Hello All, I have a question regarding Kafka Consumer Offsets. I'm currently developing a stateless processing job that reads data from a Kafka topic using the Flink DataStream Kafka Source Connector (python), but I'm not using checkpoints within the code and I have not enabled it. I also have configured the following consumer parameters: (enable.auto.commit: true, commit.offsets.on.checkpoint: false, set_starting_offsets(KafkaOffsetsInitializer.latest()). What I'm unclear about is whether the offsets are actually stored and reloaded from the Kafka broker after a Flink job restarts. With this configuration, can I guarantee that already consumed events from the topic will not be re-consumed after the restart? Besides using checkpoints, what other methods or approaches can we use to ensure that jobs do not re-consume the same records after restarts? Now on the monitoring side from what I understand, consumer lag monitoring is not possible using the classic broker lag metrics. instead, we need to use the Flink exposed metrics: (flink_taskmanager_job_task_operator_pendingRecords + flink_taskmanager_job_task_operator_KafkaSourceReader_KafkaConsumer_records_lag_max) Is this actually the correct way to monitor consumption lag when using the Flink consumer?

I think what you should concern is data loss rather than data duplication for your configuration because the offset initializer is configured to be latest. It'll start reading from the latest offsets and those that are from the last consumed offsets until the latest offsets will be skipped.

Thanks for the détails. Your right the what Inam looking to garantie is the recovery of the last consumed offsets and the to skip or start the the current latest offsets (losing of data) So if I understand correctly I must use the checkpointing for at least the Kafka source connector so that it knows where it last left off in terms of the consumed offsets? The Flink Kafka consumer documentation is really not very explicit about the pros and cons of using the built-in offsets management (checkpoints) or the native Kafka management.

I think you need to • update the offset initializer e.g.) to

KafkaOffsetsInitializer.committed_offsets(KafkaOffsetResetStrategy.EARLIEST)

• set checkpointing for fault tolerance • configure savepoints to stop/resume/update safely

Thanks for the details, so it's mandatory to have checkpoints? I did not wanted to manage and deal with checkpoints, but just relying on the offsets. Can't we just enable the enable.auto.commit?

@syepes @Jaehyeon Kim I am getting errors in pyflink 1.20.3 when I run the above recommendation... see below error. Any thoughts?

flink@d35159b00d01:~$ flink run -d -py /data/job.py --config_file /data/example-config.yaml 
Exponential restart strategy unavailable; falling back to fixed-delay restart
Traceback (most recent call last):
  File "/data/job.py", line 748, in <module>
    main(arguments.config_file)
  File "/data/job.py", line 662, in main
    .set_starting_offsets(_kafka_offsets_initializer(kafka_source_conf))
  File "/data/job.py", line 329, in _kafka_offsets_initializer
    return KafkaOffsetsInitializer.committed_offsets(KafkaOffsetResetStrategy.LATEST)
  File "/opt/flink/opt/python/pyflink.zip/pyflink/datastream/connectors/kafka.py", line 750, in committed_offsets
  File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1314, in __call__
  File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1283, in _build_args
  File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1283, in <listcomp>
  File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/protocol.py", line 298, in get_command_part
  File "/opt/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 185, in wrapped_call
TypeError: Could not found the Java class 'org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.OffsetResetStrategy.LATEST._get_object_id'. The Java dependencies could be specified via command line argument '--jarfile' or the config option 'pipeline.jars'