[ https://issues.apache.org/jira/browse/FLINK-35565?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17860998#comment-17860998 ]
dongwoo.kim commented on FLINK-35565: ------------------------------------- Since the stoppingOffset logic relies on the last record's offset, it can fall into an indefinite loop if all records are deleted. This [pr|https://github.com/apache/flink-connector-kafka/pull/100] resolves the issue. Please take a look. > Flink KafkaSource Batch Job Gets Into Infinite Loop after Resetting Offset > -------------------------------------------------------------------------- > > Key: FLINK-35565 > URL: https://issues.apache.org/jira/browse/FLINK-35565 > Project: Flink > Issue Type: Bug > Components: Connectors / Kafka > Affects Versions: kafka-3.1.0 > Environment: This is reproduced on a *Flink 1.18.1* with the latest > Kafka connector 3.1.0-1.18 on a session cluster. > Reporter: Naci Simsek > Priority: Major > Attachments: image-2024-06-11-11-19-09-889.png, > taskmanager_localhost_54489-ac092a_log.txt > > > h2. Summary > Flink batch job gets into an infinite fetch loop and could not gracefully > finish if the connected Kafka topic is empty and starting offset value in > Flink job is lower than the current start/end offset of the related topic. > See below for details: > h2. How to reproduce > Flink +*batch*+ job which works as a {*}KafkaSource{*}, will consume events > from Kafka topic. > Related Kafka topic is empty, there are no events, and the offset value is as > below: *15* > !image-2024-06-11-11-19-09-889.png|width=895,height=256! > > Flink job uses a *specific starting offset* value, which is +*less*+ than the > current offset of the topic/partition. > See below, it set as “4” > > {code:java} > package naci.grpId; > import org.apache.flink.api.common.RuntimeExecutionMode; > import org.apache.flink.api.common.serialization.SimpleStringSchema; > import org.apache.flink.connector.kafka.source.KafkaSource; > import > org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; > import org.apache.flink.streaming.api.datastream.DataStream; > import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; > import org.apache.flink.api.common.eventtime.WatermarkStrategy; > import org.apache.kafka.common.TopicPartition; > import java.util.HashMap; > import java.util.Map; > public class KafkaSource_Print { > public static void main(String[] args) throws Exception { > final StreamExecutionEnvironment env = > StreamExecutionEnvironment.getExecutionEnvironment(); > env.setRuntimeMode(RuntimeExecutionMode.BATCH); > // Define the specific offsets for the partitions > Map<TopicPartition, Long> specificOffsets = new HashMap<>(); > specificOffsets.put(new TopicPartition("topic_test", 0), 4L); // > Start from offset 4 for partition 0 > KafkaSource<String> kafkaSource = KafkaSource > .<String>builder() > .setBootstrapServers("localhost:9093") // Make sure the port > is correct > .setTopics("topic_test") > .setValueOnlyDeserializer(new SimpleStringSchema()) > > .setStartingOffsets(OffsetsInitializer.offsets(specificOffsets)) > .setBounded(OffsetsInitializer.latest()) > .build(); > DataStream<String> stream = env.fromSource( > kafkaSource, > WatermarkStrategy.noWatermarks(), > "Kafka Source" > ); > stream.print(); > env.execute("Flink KafkaSource test job"); > } > }{code} > > > Here are the initial logs printed related to the offset, as soon as the job > gets submitted: > > {code:java} > 2024-05-30 12:15:50,010 INFO > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Adding > split(s) to reader: [[Partition: topic_test-0, StartingOffset: 4, > StoppingOffset: 15]] > 2024-05-30 12:15:50,069 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Prepare to run AddSplitsTask: [[[Partition: topic_test-0, StartingOffset: 4, > StoppingOffset: 15]]] > 2024-05-30 12:15:50,074 TRACE > org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader [] - > Seeking starting offsets to specified offsets: {topic_test-0=4} > 2024-05-30 12:15:50,074 INFO org.apache.kafka.clients.consumer.KafkaConsumer > [] - [Consumer clientId=KafkaSource--2381765882724812354-0, > groupId=null] Seeking to offset 4 for partition topic_test-0 > 2024-05-30 12:15:50,075 DEBUG > org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader [] - > SplitsChange handling result: [topic_test-0, start:4, stop: 15] > 2024-05-30 12:15:50,075 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Finished running task AddSplitsTask: [[[Partition: topic_test-0, > StartingOffset: 4, StoppingOffset: 15]]] > 2024-05-30 12:15:50,075 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Prepare to run FetchTask{code} > > Since the starting offset {color:#ff0000}*4*{color} is *out of range* for the > Kafka topic, KafkaConsumer initiates an {*}offset +reset+{*}, as seen on task > manager logs: > > {code:java} > 2024-05-30 12:15:50,193 INFO > org.apache.kafka.clients.consumer.internals.Fetcher [] - [Consumer > clientId=KafkaSource--2381765882724812354-0, groupId=null] Fetch position > FetchPosition{offset=4, offsetEpoch=Optional.empty, > currentLeader=LeaderAndEpoch{leader=Optional[nacis-mbp-m2:9093 (id: 1 rack: > null)], epoch=0}} is out of range for partition topic_test-0, resetting offset > 2024-05-30 12:15:50,195 INFO > org.apache.kafka.clients.consumer.internals.SubscriptionState [] - [Consumer > clientId=KafkaSource--2381765882724812354-0, groupId=null] Resetting offset > for partition topic_test-0 to position FetchPosition{offset=15, > offsetEpoch=Optional.empty, > currentLeader=LeaderAndEpoch{leader=Optional[nacis-mbp-m2:9093 (id: 1 rack: > null)], epoch=0}}.{code} > > > Then, an {color:#ff0000}*infinite {{FetchTask}} loop*{color} starts: > > {code:java} > 2024-05-30 12:16:00,079 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Finished running task FetchTask > 2024-05-30 12:16:00,079 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Prepare to run FetchTask > 2024-05-30 12:16:00,079 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Current > fetch is finished. > 2024-05-30 12:16:00,080 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Source > reader status: NOTHING_AVAILABLE > 2024-05-30 12:16:06,288 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from df54e7abdfa0095dc5c214b056153dea. > 2024-05-30 12:16:08,755 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from e1746de110bfdd23c7dba50f3b083621. > 2024-05-30 12:16:10,082 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Finished running task FetchTask > 2024-05-30 12:16:10,082 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Prepare to run FetchTask > 2024-05-30 12:16:10,082 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Getting > next source data batch from queue > 2024-05-30 12:16:10,082 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Current > fetch is finished. > 2024-05-30 12:16:10,082 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Source > reader status: NOTHING_AVAILABLE > 2024-05-30 12:16:16,290 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from df54e7abdfa0095dc5c214b056153dea. > 2024-05-30 12:16:17,393 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > file upload request for file LOG > 2024-05-30 12:16:17,394 DEBUG org.apache.flink.runtime.blob.BlobClient > [] - PUT BLOB stream to /127.0.0.1:55663. > 2024-05-30 12:16:18,757 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from e1746de110bfdd23c7dba50f3b083621. > 2024-05-30 12:16:20,084 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Finished running task FetchTask > 2024-05-30 12:16:20,084 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Getting > next source data batch from queue > 2024-05-30 12:16:20,084 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Prepare to run FetchTask > 2024-05-30 12:16:20,084 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Current > fetch is finished. > 2024-05-30 12:16:20,084 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Source > reader status: NOTHING_AVAILABLE > 2024-05-30 12:16:26,293 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from df54e7abdfa0095dc5c214b056153dea. > 2024-05-30 12:16:28,761 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from e1746de110bfdd23c7dba50f3b083621. > 2024-05-30 12:16:30,086 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Finished running task FetchTask > 2024-05-30 12:16:30,086 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Getting > next source data batch from queue > 2024-05-30 12:16:30,086 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Prepare to run FetchTask > 2024-05-30 12:16:30,086 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Current > fetch is finished. > 2024-05-30 12:16:30,086 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Source > reader status: NOTHING_AVAILABLE > 2024-05-30 12:16:36,296 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from df54e7abdfa0095dc5c214b056153dea. > 2024-05-30 12:16:38,762 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from e1746de110bfdd23c7dba50f3b083621. > 2024-05-30 12:16:40,087 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Finished running task FetchTask > 2024-05-30 12:16:40,087 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Getting > next source data batch from queue > 2024-05-30 12:16:40,087 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Prepare to run FetchTask > 2024-05-30 12:16:40,088 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Current > fetch is finished. > 2024-05-30 12:16:40,088 TRACE > org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Source > reader status: NOTHING_AVAILABLE > 2024-05-30 12:16:46,297 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from df54e7abdfa0095dc5c214b056153dea. > 2024-05-30 12:16:48,765 DEBUG > org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received > heartbeat request from e1746de110bfdd23c7dba50f3b083621. > 2024-05-30 12:16:50,089 DEBUG > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - > Finished running task FetchTask{code} > > The loop *ends* as soon as I *add* a *new event* on this Kafka topic, which > will be placed in offset 15. > {+}*Expected Result*{+}: Since this is a batch job, and since there is no > event on the Kafka topic, right after offset reset, Flink connector should > identify that there is no events to process, and gracefully finish the > application. > {+}*Actual Result*{+}: Flink connector infinitely tries to fetch an event > from offset:15 which actually exists but no events on that offset, > application keep fetching that same offset! > > This issue is +*NOT*+ happening if the above Flink application sets a > *starting offset* +*15*+ or {+}*higher*{+}! If it is given as 15 or higher, > no offset reset is performed, and the Flink application gracefully finishes! > This is reproduced on a *Flink 1.18.1* with the latest Kafka connector > 3.1.0-1.18 on a session cluster. > Logs are attached. -- This message was sent by Atlassian Jira (v8.20.10#820010)