[jira] [Created] (FLINK-34470) Transactional message + Table api kafka source with 'latest-offset' scan bound mode causes indefinitely hanging

["dongwoo.kim (Jira)" <ji...@apache.org>]

dongwoo.kim created FLINK-34470:
-----------------------------------

             Summary: Transactional message + Table api kafka source with 'latest-offset' scan bound mode causes indefinitely hanging
                 Key: FLINK-34470
                 URL: https://issues.apache.org/jira/browse/FLINK-34470
             Project: Flink
          Issue Type: Bug
          Components: Connectors / Kafka
    Affects Versions: 1.17.1
            Reporter: dongwoo.kim


h2. Summary  
Hi we have faced issue with transactional message and table api kafka source. 
If we configure *'scan.bounded.mode'* to *'latest-offset'* flink sql request timeouts after  hanging. We can always reproduce this unexpected behavior by following below steps.


h2.  How to reproduce
1. Deploy transactional producer and stop after producing certain amount of messages
2. Configure *'scan.bounded.mode'* to *'latest-offset'* and submit simple query such as count(*)
3. Flink sql job gets stucked and timeouts.

h2. Cause
Transaction producer always produces [control records|https://kafka.apache.org/documentation/#controlbatch] at the end of the transaction. And these controll messages are not polled by {*}consumer.poll(){*}. (It is filtered internally). In *KafkaPartitionSplitReader* code it finishes split only when 
*lastRecord.offset() >= stoppingOffset - 1* condition is met. This might work well with non transactional messages or streaming environment but in some batch use cases it causes unexpected hanging.

h2. Proposed solution
Adding *consumer.position(tp) >= stoppingOffset* condition to the if statement. 
By this KafkaPartitionSplitReader is available to finish the split even when the stopping offset is configured to control record's offset. 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)