public KafkaFetcher()

in flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaFetcher.java [78:118]

39 lines of code
1 McCabe index (conditional complexity)


    public KafkaFetcher(
            SourceFunction.SourceContext<T> sourceContext,
            Map<KafkaTopicPartition, Long> assignedPartitionsWithInitialOffsets,
            SerializedValue<WatermarkStrategy<T>> watermarkStrategy,
            ProcessingTimeService processingTimeProvider,
            long autoWatermarkInterval,
            ClassLoader userCodeClassLoader,
            String taskNameWithSubtasks,
            KafkaDeserializationSchema<T> deserializer,
            Properties kafkaProperties,
            long pollTimeout,
            MetricGroup subtaskMetricGroup,
            MetricGroup consumerMetricGroup,
            boolean useMetrics)
            throws Exception {
        super(
                sourceContext,
                assignedPartitionsWithInitialOffsets,
                watermarkStrategy,
                processingTimeProvider,
                autoWatermarkInterval,
                userCodeClassLoader,
                consumerMetricGroup,
                useMetrics);

        this.deserializer = deserializer;
        this.handover = new Handover();

        this.consumerThread =
                new KafkaConsumerThread(
                        LOG,
                        handover,
                        kafkaProperties,
                        unassignedPartitionsQueue,
                        getFetcherName() + " for " + taskNameWithSubtasks,
                        pollTimeout,
                        useMetrics,
                        consumerMetricGroup,
                        subtaskMetricGroup);
        this.kafkaCollector = new KafkaCollector();
    }