@@ -0,0 +1,3390 @@
+package org.talend.components.cassandra.io.bd;%0A%0Aimport com.google.cloud.dataflow.sdk.coders.Coder;%0Aimport com.google.cloud.dataflow.sdk.io.BoundedSource;%0Aimport com.google.cloud.dataflow.sdk.options.DataflowPipelineWorkerPoolOptions;%0Aimport com.google.cloud.dataflow.sdk.options.PipelineOptions;%0Aimport org.talend.components.api.component.runtime.input.SingleSplit;%0Aimport org.talend.components.api.component.runtime.input.Source;%0Aimport org.talend.components.api.component.runtime.input.Split;%0Aimport org.talend.components.api.properties.ComponentProperties;%0A%0Aimport java.io.IOException;%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0Aimport java.util.NoSuchElementException;%0A%0A/**%0A * Created by bchen on 16-1-17.%0A */%0Apublic class DFBoundedSource extends BoundedSource %7B%0A    Source source;%0A    Split split;%0A%0A    public DFBoundedSource(Class%3C? extends Source%3E sourceClazz, ComponentProperties props) %7B%0A        try %7B%0A            this.source = sourceClazz.newInstance();%0A        %7D catch (InstantiationException e) %7B%0A            e.printStackTrace();%0A        %7D catch (IllegalAccessException e) %7B%0A            e.printStackTrace();%0A        %7D%0A        this.source.init(props);%0A    %7D%0A%0A    public DFBoundedSource(Split split) %7B%0A        this.split = split;%0A    %7D%0A%0A    @Override%0A    public List%3C? extends BoundedSource%3E splitIntoBundles(long desiredBundleSizeBytes, PipelineOptions options) throws Exception %7B%0A        List%3CDFBoundedSource%3E sourceList = new ArrayList%3C%3E();%0A        if (source.supportSplit()) %7B%0A            DataflowPipelineWorkerPoolOptions poolOptions =%0A                    options.as(DataflowPipelineWorkerPoolOptions.class);%0A            Split%5B%5D split = source.getSplit(poolOptions.getNumWorkers());%0A            for (Split s : split) %7B%0A                sourceList.add(new DFBoundedSource(s));%0A            %7D%0A        %7D else %7B%0A            sourceList.add(new DFBoundedSource(new SingleSplit()));%0A        %7D%0A        return sourceList;%0A    %7D%0A%0A    @Override%0A    public long getEstimatedSizeBytes(PipelineOptions options) throws Exception %7B%0A        //TODO source.getCount?%0A        return 0;%0A    %7D%0A%0A    @Override%0A    public boolean producesSortedKeys(PipelineOptions options) throws Exception %7B%0A        return false;%0A    %7D%0A%0A    @Override%0A    public BoundedReader createReader(PipelineOptions options) throws IOException %7B%0A        return new DFBoundedReader(source.getRecordReader(split));%0A    %7D%0A%0A    @Override%0A    public void validate() %7B%0A%0A    %7D%0A%0A    @Override%0A    public Coder getDefaultOutputCoder() %7B%0A        return null;%0A    %7D%0A%0A    public class DFBoundedReader extends BoundedReader %7B%0A        org.talend.components.api.component.runtime.input.Reader reader;%0A%0A        public DFBoundedReader(org.talend.components.api.component.runtime.input.Reader reader) %7B%0A            this.reader = reader;%0A        %7D%0A%0A        @Override%0A        public boolean start() throws IOException %7B%0A            return reader.start();%0A        %7D%0A%0A        @Override%0A        public boolean advance() throws IOException %7B%0A            return reader.advance();%0A        %7D%0A%0A        @Override%0A        public Object getCurrent() throws NoSuchElementException %7B%0A            return reader.getCurrent();%0A        %7D%0A%0A        @Override%0A        public void close() throws IOException %7B%0A            reader.close();%0A        %7D%0A%0A        @Override%0A        public BoundedSource getCurrentSource() %7B%0A            return null;%0A        %7D%0A    %7D%0A%7D%0A

add BoundedSource for Dataflow sdk