Kafka 4.x Queue Semantics support

[Shekharrajak]

Ref https://issues.apache.org/jira/browse/FLINK-38287

This implementation adds Kafka 4.x share group semantics to Flink's Kafka connector while maintaining full backward compatibility with existing code. The code changes are following KIP-932 and FLIP-27 main source architecture and implicit mode acknowledgement.

This directly addresses use cases where:

1. Multiple consumers need to process items efficiently in parallel from a single/multiple topic(s).
2. Messages need explicit acknowledgment/release (to avoid reprocessing or allow retries).
   Use cases where scaling Flink ML/LLM workload is critical - Shifting Kafka coordination and assignment logic to the broker side would simplify today’s complex Flink source management, making consumption more efficient, scalable, and far less error-prone.
   Operational Benefits

• Higher Throughput: ShareGroupHeartbeat helps in Queue-like workloads, maximum throughput scenarios. Share groups distribute messages at the record level, not partition level, so multiple readers can consume from the same topic with Kafka coordinating message distribution.
• Better Availability and Flexible Scaling: consumers assignment logic is simpler in server side and rebalancing frequency is minimised.

Note:

Semantic Guarantee
AT-MOST-ONCE: Records acknowledged immediately after polling. Job failure after acknowledgment but
before checkpoint = permanent data loss.

Ref https://issues.apache.org/jira/browse/KAFKA-19883 implemented for no data loss in commit : ef83f90

Good for: Logs, monitoring, non-critical analyticsNot for: Financial transactions, critical data,
audit trails

[boring-cyborg]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[Shekharrajak]

These changes can be reverted.

[Shekharrajak]

Testcases improvements is required, I will check and update them accordingly.

[Shekharrajak]

This is not yet released and expected to be available to download. Meantime the testing is done by adding this into class path.

[Shekharrajak]

ListCheckpointed will help to store the records which is polled but not yet processed in Flink persistent checkpoint state - this will make sure in case of any failure / crash we process the records that we read & ack.

[Shekharrajak]

Using Flink SQL, we can have some validation :

CREATE TABLE kafka_share_source (
      message STRING
  ) WITH (
      'connector' = 'kafka-sharegroup',
      'bootstrap.servers' = 'localhost:9092',
      'share-group-id' = 'flink-sql-test-group',
      'topic' = 'test-topic',
      'format' = 'raw',
      'source.parallelism' = '4'  -- 4 subtasks regardless of partition count
  );
 
select * from kafka_share_source;

[davidradl]

PR 190 is also upping the Kafka client level to 4.1. If we do it here then we should amend the NOTOCE as per pr 190. fyi @tomncooper

[jnh5y]

As a high-level note, since share groups do not use transactions, there will be some possibility for reprocessing messages. Is that ok for your use cases?

Generally, do you have any performance numbers to show that this consumer is faster? (Of course, since transactions are not available, I could imagine it being a little bit faster anyhow...)

[Shekharrajak]

Flink-SQL:

 //  {"order_id":1001,"customer_id":"C001","quantity":2,"price":25.50}
   CREATE TABLE orders_share_group (
      order_id BIGINT,
      customer_id STRING,
      quantity INT,
      price DECIMAL(10, 2)
  ) WITH (
      'connector' = 'kafka-sharegroup',
      'topic' = 'orders',
      'bootstrap.servers' = 'localhost:9092',
      'share-group-id' = 'flink-orders-sharegroup',
      'format' = 'json'

  );

  select * from orders_share_group;