@Override
    protected AbstractFetcher<Row, ?> createFetcher(SourceFunction.SourceContext<Row> sourceContext,
                                                    Map<KafkaTopicPartition, Long> assignedPartitionsWithInitialOffsets,
                                                    SerializedValue<AssignerWithPeriodicWatermarks<Row>> watermarksPeriodic,
                                                    SerializedValue<AssignerWithPunctuatedWatermarks<Row>> watermarksPunctuated,
                                                    StreamingRuntimeContext runtimeContext,
                                                    OffsetCommitMode offsetCommitMode,
                                                    MetricGroup consumerMetricGroup,
                                                    boolean useMetrics) throws Exception {

        final OffsetMap offsetMap = sampleSize > 0 ? seekOffset(props, topic) : new OffsetMap();
        Map<KafkaTopicPartition, Long> rebuild;

        rebuild =
                sampleSize > 0
                        ? SampleCalculateHelper.rebuildAssignedPartitionsWithInitialOffsets(
                                offsetMap, sampleSize, assignedPartitionsWithInitialOffsets)
                        : assignedPartitionsWithInitialOffsets;

        AbstractFetcher<Row, ?> fetcher =
                super.createFetcher(
                        sourceContext,
                        rebuild,
                        watermarksPeriodic,
                        watermarksPunctuated,
                        runtimeContext,
                        offsetCommitMode,
                        consumerMetricGroup,
                        useMetrics);
        ((KafkaDeserializationMetricWrapper) deserializationMetricWrapper).setFetcher(fetcher);
        ((DtKafkaDeserializationSchemaWrapper<?>) deserializer).setSpecificEndOffsets(offsetMap);
        return fetcher;
    }
Reason:
Date: 2018/10/19
Company: www.dtstack.com
@author xuchao
kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer.java