[ https://issues.apache.org/jira/browse/FLINK-35506?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
elon_X updated FLINK-35506: --------------------------- Description: When I use KafkaSource for consuming topics and set the Kafka parameter {{{}enable.auto.commit=true{}}}, while also enabling checkpointing for the task, I notice that both will commit offsets. Should Kafka's auto-commit be disabled when enabling Flink checkpointing, similar to how it's done with FlinkKafkaConsumer? *How to reproduce* {code:java} // code placeholder Properties kafkaParams = new Properties(); kafkaParams.put("enable.auto.commit", "true"); kafkaParams.put("auto.offset.reset", "latest"); kafkaParams.put("fetch.min.bytes", "4096"); kafkaParams.put("sasl.mechanism", "PLAIN"); kafkaParams.put("security.protocol", "SASL_PLAINTEXT"); kafkaParams.put("bootstrap.servers", bootStrap); kafkaParams.put("group.id", expoGroupId); kafkaParams.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";"); KafkaSource<String> source = KafkaSource .<String>builder() .setBootstrapServers(bootStrap) .setProperties(kafkaParams) .setGroupId(expoGroupId) .setTopics(Arrays.asList(expoTopic)) .setValueOnlyDeserializer(new SimpleStringSchema()) .setStartingOffsets(OffsetsInitializer.latest()) .build(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source") .filter(r -> true); env.enableCheckpointing(3000 * 1000); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000 * 1000); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE); env.getCheckpointConfig().setCheckpointTimeout(1000 * 300); env.execute("kafka-consumer"); {code} 'org.apache.kafka.clients.consumer.internals.ConsumerCoordinator' continuously committing offsets. !image-2024-06-03-23-39-28-270.png! was:When I use KafkaSource for consuming topics and set the Kafka parameter {{{}enable.auto.commit=true{}}}, while also enabling checkpointing for the task, I notice that both will commit offsets. Should Kafka's auto-commit be disabled when enabling Flink checkpointing, similar to how it's done with FlinkKafkaConsumer? > disable kafka auto-commit and rely on flink’s checkpointing if both are > enabled > ------------------------------------------------------------------------------- > > Key: FLINK-35506 > URL: https://issues.apache.org/jira/browse/FLINK-35506 > Project: Flink > Issue Type: Improvement > Components: Connectors / Kafka > Affects Versions: 1.16.1 > Reporter: elon_X > Priority: Major > Attachments: image-2024-06-03-23-39-28-270.png > > > When I use KafkaSource for consuming topics and set the Kafka parameter > {{{}enable.auto.commit=true{}}}, while also enabling checkpointing for the > task, I notice that both will commit offsets. Should Kafka's auto-commit be > disabled when enabling Flink checkpointing, similar to how it's done with > FlinkKafkaConsumer? > > *How to reproduce* > > {code:java} > // code placeholder > Properties kafkaParams = new Properties(); > kafkaParams.put("enable.auto.commit", "true"); > kafkaParams.put("auto.offset.reset", "latest"); > kafkaParams.put("fetch.min.bytes", "4096"); > kafkaParams.put("sasl.mechanism", "PLAIN"); > kafkaParams.put("security.protocol", "SASL_PLAINTEXT"); > kafkaParams.put("bootstrap.servers", bootStrap); > kafkaParams.put("group.id", expoGroupId); > kafkaParams.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule > required username=\"" + username + "\" password=\"" + password + "\";"); > KafkaSource<String> source = KafkaSource > .<String>builder() > .setBootstrapServers(bootStrap) > .setProperties(kafkaParams) > .setGroupId(expoGroupId) > .setTopics(Arrays.asList(expoTopic)) > .setValueOnlyDeserializer(new SimpleStringSchema()) > .setStartingOffsets(OffsetsInitializer.latest()) > .build(); > StreamExecutionEnvironment env = > StreamExecutionEnvironment.getExecutionEnvironment(); > env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source") > .filter(r -> true); > env.enableCheckpointing(3000 * 1000); > env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000 * 1000); > env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE); > env.getCheckpointConfig().setCheckpointTimeout(1000 * 300); > env.execute("kafka-consumer"); {code} > > > 'org.apache.kafka.clients.consumer.internals.ConsumerCoordinator' > continuously committing offsets. > !image-2024-06-03-23-39-28-270.png! -- This message was sent by Atlassian Jira (v8.20.10#820010)