feat: [Phase 1 - Draft] Queue Semantics support in Kafka Ingestion by Shekharrajak · Pull Request #19311 · apache/druid

Shekharrajak · 2026-04-14T16:43:37Z

Phase 1 #18439.

Description

Fixed the bug ...

Renamed the class ...

Added a forbidden-apis entry ...

Release note

Key changed/added classes in this PR

MyFoo
OurBar
TheirBaz

This PR has:

Demo

https://youtu.be/K_O1MH-AaE8

Shekharrajak · 2026-04-15T15:13:02Z

+cd apache-druid-31.0.0
+
+# Replace the kafka extension with our build
+rm extensions/druid-kafka-indexing-service/*.jar


We can build from the source and use the tar

Shekharrajak · 2026-04-15T15:15:11Z

+        agg -> agg.hasSumAtLeast(numRecords)
+    );
+
+    Assertions.assertEquals(


Making sure we read all the records.

Shekharrajak · 2026-05-08T06:41:18Z

Spending time on Integeration testing embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/EmbeddedShareGroupIngestionTest.java

Shekharrajak · 2026-05-08T07:13:25Z

For smaller blast radius and clean separation - parking the suggestion #18439 (comment) , Also KafkaShareGroupSupervisor (planned) will be structurally different from KafkaSupervisor — no partition-to-task assignment, no offset bookkeeping, seek/rest .

Shekharrajak · 2026-05-08T07:22:38Z

+  }
+
+  @Test
+  public void test_shareGroupIngestion_basicEndToEnd() throws InterruptedException


Subscribe to topic with KafkaShareConsumer (groupId), poll → parse (CSV) → persist → publish → acknowledge(ACCEPT) → commitSync → broker offset persistence. SQL COUNT(*) == 10.

Shekharrajak · 2026-05-08T07:24:52Z

+    final int numRecords = 5;
+    final int rowsPerRecord = 4;
+    kafkaServer.produceRecordsToTopic(
+        generateMultiObjectJsonRecords(topic, numRecords, rowsPerRecord, DateTimes.of("2025-07-01"))


StreamChunkReader multi-row path One Kafka record carrying 4 whitespace-separated JSON objects produces 4 rows. SQL count = numRecords × rowsPerRecord (5 × 4 = 20). Verifies no row is dropped before ACK.

Shekharrajak · 2026-05-08T07:30:29Z

To make sure no data loss we are ack only after the publish completed https://github.com/apache/druid/pull/19311/changes#diff-487176908fb14d87486db6f683c9ddb34465f3a1572f5c01b11d7b9c6a482289R381

Although this will be slow and no concurrency but I think for the intial phase this is fine and we can improve the performance step by step.

WDYT @gianm @a2l007 @clintropolis @FrankChen021 ?

Shekharrajak · 2026-05-08T17:54:12Z

CI checks are failing due to one flaky test unrelated to this PR : #19435

+    recordMap.put(new TopicPartition("test-topic", 0), recordsP0);
+    recordMap.put(new TopicPartition("test-topic", 1), recordsP1);
+    Mockito.when(mockConsumer.poll(Mockito.any(Duration.class)))
+           .thenReturn(new ConsumerRecords<>(recordMap));


+    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>();
+    recordMap.put(new TopicPartition(record.topic(), record.partition()), List.of(record));
+    Mockito.when(mockConsumer.poll(Mockito.any(Duration.class)))
+           .thenReturn(new ConsumerRecords<>(recordMap));


+   */
+  Map<PartitionIdType, Optional<Exception>> commitSync();
+
+  Set<PartitionIdType> getPartitionIds(String stream);


FrankChen021

Severity	Findings
P0	0
P1	0
P2	2
P3	0
Total	2

Reviewed 22 of 22 changed files. Findings focus on task scheduling semantics and Kafka compatibility test coverage.

This is an automated review by Codex GPT-5

FrankChen021 · 2026-05-09T05:39:16Z

+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <excludes>
+            <exclude>**/KafkaIndexTaskTest.java</exclude>


[P2] Kafka compatibility tests are skipped globally

This Surefire exclusion removes the main KafkaIndexTask, KafkaRecordSupplier, KafkaSamplerSpec, and KafkaSupervisor test classes from every normal test run. Since this PR also upgrades kafka-clients to 4.2.0, those are the regression tests most likely to catch compatibility breaks in existing index_kafka ingestion; the new share-group mock and embedded tests do not cover that existing API surface. Please migrate or narrowly isolate the TestBroker-dependent cases instead of excluding the whole classes.

Yes, planning to fix (upgrade to 4.2 version in all places) as part of this #19322 - I feel that we need bunch of changes, API compatibility issues in running tests , kafka docker image for IT tests and in source code - so want to keep it in separate PR.

This PR mainly focus on Share group implementation.

Shekharrajak · 2026-05-09T06:13:01Z

Struggling with CI check related to license :


Unsupported license: MPL 1.1
For:org.javassist javassist in: druid-kerberos
org.locationtech.jts jts-core in: parquet-extensions with: Eclipse Public License 2.0 ignoring Eclipse Distribution License - v 1.0
Error:  Command execution failed.
org.apache.commons.exec.ExecuteException: Process exited with an error: 1 (Exit value: 1)
    at org.apache.commons.exec.DefaultExecutor.executeInternal (DefaultExecutor.java:394)
    at org.apache.commons.exec.DefaultExecutor.execute (DefaultExecutor.java:292)
    at org.codehaus.mojo.exec.ExecMojo.executeCommandLine (ExecMojo.java:903)
    at org.codehaus.mojo.exec.ExecMojo.executeCommandLine (ExecMojo.java:863)
    at org.codehaus.mojo.exec.ExecMojo.execute (ExecMojo.java:454)
    at org.apache.maven.plugin.DefaultBuildPluginManager.executeMojo (DefaultBuildPluginManager.java:126)
    at org.apache.maven.lifecycle.internal.MojoExecutor.doExecute2 (MojoExecutor.java:328)
    at org.apache.maven.lifecycle.internal.MojoExecutor.doExecute (MojoExecutor.java:316)
    at org.apache.maven.lifecycle.internal.MojoExecutor.execute (MojoExecutor.java:212)
    at org.apache.maven.lifecycle.internal.MojoExecutor.execute (MojoExecutor.java:174)
    at org.apache.maven.lifecycle.internal.MojoExecutor.access$000 (MojoExecutor.java:75)
    at org.apache.maven.lifecycle.internal.MojoExecutor$1.run (MojoExecutor.java:162)
    at org.apache.maven.plugin.DefaultMojosExecutionStrategy.execute (DefaultMojosExecutionStrategy.java:39)
    at org.apache.maven.lifecycle.internal.MojoExecutor.execute (MojoExecutor.java:159)
    at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject (LifecycleModuleBuilder.java:105)
    at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject (LifecycleModuleBuilder.java:73)
    at org.apache.maven.lifecycle.internal.builder.singlethreaded.SingleThreadedBuilder.build (SingleThreadedBuilder.java:53)
    at org.apache.maven.lifecycle.internal.LifecycleStarter.execute (LifecycleStarter.java:118)
    at org.apache.maven.DefaultMaven.doExecute (DefaultMaven.java:261)
[INFO] simple-client-sslcontext ........................... SUCCESS [  5.195 s]
[INFO] druid-basic-security ............................... SUCCESS [  9.326 s]
[INFO] druid-google-extensions ............................ SUCCESS [  7.697 s]
[INFO] druid-catalog ...................................... SUCCESS [  9.519 s]
[INFO] druid-testing-tools ................................ SUCCESS [  8.122 s]
[INFO] druid-testcontainers ............................... SUCCESS [  2.149 s]
[INFO] druid-influx-extensions ............................ SUCCESS [  4.579 s]
[INFO] druid-cassandra-storage ............................ SUCCESS [  3.987 s]
[INFO] dropwizard-emitter ................................. SUCCESS [  6.528 s]
[INFO] druid-cloudfiles-extensions ........................ SUCCESS [  5.857 s]
[INFO] druid-consul-extensions ............................ SUCCESS [  6.595 s]
[INFO] graphite-emitter ................................... SUCCESS [  6.085 s]
[INFO] druid-distinctcount ................................ SUCCESS [  4.219 s]
[INFO] druid-exact-count-bitmap ........................... SUCCESS [  7.275 s]
[INFO] statsd-emitter ..................................... SUCCESS [  6.282 s]
[INFO] druid-time-min-max ................................. SUCCESS [  5.725 s]
[INFO] druid-virtual-columns .............................. SUCCESS [  4.011 s]
[INFO] druid-thrift-extensions ............................ SUCCESS [  4.915 s]
[INFO] ambari-metrics-emitter ............................. SUCCESS [  6.306 s]
[INFO] sqlserver-metadata-storage ......................... SUCCESS [  6.266 s]
[INFO] kafka-emitter ...................................... SUCCESS [  6.186 s]
[INFO] druid-redis-cache .................................. SUCCESS [  7.346 s]
[INFO] druid-opentsdb-emitter ............................. SUCCESS [  5.524 s]
[INFO] druid-momentsketch ................................. SUCCESS [  6.063 s]
[INFO] druid-moving-average-query ......................... SUCCESS [  7.560 s]
[INFO] tdigestsketch ...................................... SUCCESS [  7.308 s]
[INFO] druid-ddsketch ..................................... SUCCESS [  7.400 s]
[INFO] druid-influxdb-emitter ............................. SUCCESS [  4.017 s]
[INFO] gce-extensions ..................................... SUCCESS [  6.621 s]
[INFO] aliyun-oss-extensions .............................. SUCCESS [  7.980 s]
[INFO] prometheus-emitter ................................. SUCCESS [  5.877 s]
[INFO] opentelemetry-emitter .............................. SUCCESS [  8.167 s]
[INFO] grpc-query ......................................... SUCCESS [ 14.755 s]
[INFO] druid-iceberg-extensions ........................... SUCCESS [  9.575 s]
[INFO] druid-deltalake-extensions ......................... SUCCESS [  6.163 s]
[INFO] druid-spectator-histogram .......................... SUCCESS [  8.500 s]
[INFO] druid-rabbit-indexing-service ...................... SUCCESS [  7.905 s]
[INFO] druid-ranger-security .............................. SUCCESS [  9.363 s]
[INFO] distribution ....................................... FAILURE [05:29 min]
[INFO] druid-quidem-ut .................................... SKIPPED
[INFO] druid-embedded-tests ............................... SKIPPED
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  22:30 min
[INFO] Finished at: 2026-05-09T05:59:40Z
[INFO] ------------------------------------------------------------------------
Error:  Failed to execute goal org.codehaus.mojo:exec-maven-plugin:3.5.1:exec (check-licenses) on project distribution: Command execution failed. Process exited with an error: 1 (Exit value: 1) -> [Help 1]
Error:  
Error:  To see the full stack trace of the errors, re-run Maven with the -e switch.
Error:  Re-run Maven using the -X switch to enable full debug logging.
Error:  
Error:  For more information about the errors and possible solutions, please read the following articles:
Error:  [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException
Error:  
Error:  After correcting the problems, you can resume the build with the command
Error:    mvn <args> -rf :distribution
Error: Process completed with exit code 1.

FrankChen021

Reviewed 3 of 27 changed files. Follow-up review handled: the Kafka extension POM no longer has the global Surefire exclusions, and ShareGroupIndexTask now uses DEFAULT_REALTIME_TASK_PRIORITY with tests covering the default and context override cases.

This is an automated review by Codex GPT-5.5

…r queue-semantics ingestion

…re group ingestion

…plicit acknowledgement

…lient to 4.2.0

…d-tests

…ptions

… bottom

FrankChen021

Severity	Findings
P0	0
P1	0
P2	2
P3	0
Total	2

Reviewed 21 of 21 changed files.

This is an automated review by Codex GPT-5.5

… share-group test

…vider expansion

…ing demo records

+    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>();
+    recordMap.put(new TopicPartition("test-topic", 0), List.of(record1));
+    recordMap.put(new TopicPartition("test-topic", 1), List.of(record2));
+    final ConsumerRecords<byte[], byte[]> consumerRecords = new ConsumerRecords<>(recordMap);


+    );
+    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>();
+    recordMap.put(new TopicPartition("test-topic", 0), List.of(record));
+    when(mockConsumer.poll(any(Duration.class))).thenReturn(new ConsumerRecords<>(recordMap));


…OrderedPartitionableRecord data

FrankChen021

Follow-up handled. Reviewed 21 of 21 changed files. The demo now configures share.auto.offset.reset=earliest before producing sample records, and share-group consumer properties are sanitized again after dynamic config expansion. I do not have further inline replies on the two follow-up threads.

This is an automated review by Codex GPT-5.5

Shekharrajak · 2026-05-15T17:26:51Z

CI Check error - unrelated timeouts :

  embedded-tests/target/test-classes/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.class:282 | Timed out waiting for event after [20,000]ms 20.391s
  embedded-tests/target/test-classes/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.class:337 | Timed out waiting for event after [20,000]ms 20.015s
  embedded-tests/target/test-classes/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.class:166 | Timed out waiting for event after [20,000]ms 20.019s
  ℹ️ Posting with conclusion 'failure' to https://github.com/apache/druid/pull/19311 (sha: 2d094d5fbe6e11b6d22b1df8bf54c8cf7b7d03d2)
🚀 Publish results
  ℹ️ - Test Report - test-jdk25-[I*,A*,U*] - 18604 tests run, 17847 passed, 754 skipped, 3 failed.
     🧪 - embedded-tests/target/test-classes/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.class | Timed out waiting for event after [20,000]ms
     🧪 - embedded-tests/target/test-classes/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.class | Timed out waiting for event after [20,000]ms
     🧪 - embedded-tests/target/test-classes/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.class | Timed out waiting for event after [20,000]ms

Raised ticket #19463

Shekharrajak · 2026-05-15T17:34:55Z

Another flaky test: #19464

github-actions Bot added Area - Documentation Area - Streaming Ingestion Area - Dependencies Area - Ingestion labels Apr 14, 2026

Shekharrajak mentioned this pull request Apr 14, 2026

Kafka 4.x Queue Semantics support in Kafka Ingestion #18439

Open

Shekharrajak commented Apr 15, 2026

View reviewed changes

FrankChen021 reviewed Apr 25, 2026

View reviewed changes

Comment thread ...ndexing-service/src/main/java/org/apache/druid/indexing/kafka/ShareGroupIndexTaskRunner.java Outdated

Shekharrajak force-pushed the feature/kafka-share-group-ingestion branch from d31199b to 4a30bee Compare May 8, 2026 06:17

Shekharrajak changed the title ~~[DRAFT] Queue Semantics support in Kafka Ingestion~~ feat: [Phase 1 - Draft] Queue Semantics support in Kafka Ingestion May 8, 2026

Shekharrajak commented May 8, 2026

View reviewed changes

github-advanced-security AI found potential problems May 8, 2026

View reviewed changes

Shekharrajak force-pushed the feature/kafka-share-group-ingestion branch from 234a8ec to e5894ef Compare May 9, 2026 05:36

FrankChen021 reviewed May 9, 2026

View reviewed changes

Shekharrajak force-pushed the feature/kafka-share-group-ingestion branch 2 times, most recently from e3a9ad7 to 7865104 Compare May 9, 2026 06:57

Shekharrajak mentioned this pull request May 9, 2026

build(deps): Kafka client 4.2.0 upgrade #19441

Merged

9 tasks

FrankChen021 reviewed May 10, 2026

View reviewed changes

Shekharrajak added 5 commits May 13, 2026 23:04

Add AcknowledgingRecordSupplier interface and AcknowledgeType enum fo…

3d9735c

…r queue-semantics ingestion

Add ShareGroupIndexTask and ShareGroupIndexTaskIOConfig for Kafka sha…

8351a82

…re group ingestion

Add KafkaShareGroupRecordSupplier wrapping KafkaShareConsumer with ex…

ac827e7

…plicit acknowledgement

Add ShareGroupIndexTaskRunner with poll-parse-publish-ack ingestion loop

da77a47

Register ShareGroupIndexTask in KafkaIndexTaskModule and bump Kafka c…

10fa462

…lient to 4.2.0

Shekharrajak force-pushed the feature/kafka-share-group-ingestion branch from ede5072 to e159405 Compare May 13, 2026 17:45

Shekharrajak added 13 commits May 13, 2026 23:39

Tighten seekablestream common javadocs

7d7bd13

Update share-group unit tests for new ack contract

fd09eca

Move share-group ITs to embedded-tests and add probe test

bc61817

Pin kafka-clients 4.2.0 in embedded-tests

210cd04

Update docs for embedded-tests share-group IT location

2356d65

Trim share-group code comments to essentials

d2e94de

Tighten share-group docs and fix spellcheck

446e004

Wait for segments before SQL in share-group IT

461b50b

Retry SQL row count to absorb broker catalog refresh lag

1226944

fix: checkstyle violations in share-group sources and tests

52dfecc

fix: forbidden-API violations in share-group code

0b53ae7

ShareGroupIndexTask: use realtime task priority (75)

d2882c0

chore(test): use ${apache.kafka.version} for kafka-clients in embedde…

1a8bed2

…d-tests

Shekharrajak force-pushed the feature/kafka-share-group-ingestion branch from e159405 to 1a8bed2 Compare May 13, 2026 18:13

Shekharrajak added 6 commits May 13, 2026 23:46

docs: drop Phase 1/Phase 2 phrasing in share-group ingestion doc

244ffde

docs: drop remaining Phase 1/Phase 2 phrasing in tuningConfig section

6a25c78

docs: drop Phase 1 phrasing in tuningConfig intro

5231b7c

docs(share-group): replace download steps with source build for the demo

40d5ed7

docs(share-group): offer source-build (recommended) and JAR-overlay o…

626b468

…ptions

fix(checkstyle): drop unused mockito imports and move static block to…

f182aae

… bottom

FrankChen021 reviewed May 14, 2026

View reviewed changes

Comment thread docs/ingestion/kafka-share-group-ingestion.md

Comment thread ...ing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaShareGroupRecordSupplier.java

Shekharrajak added 4 commits May 14, 2026 21:25

fix(forbidden-api): pass StandardCharsets.UTF_8 to String.getBytes in…

a25b7e7

… share-group test

fix(share-group): re-sanitise consumer props after dynamic-config pro…

539f3c8

…vider expansion

docs(share-group): set share.auto.offset.reset=earliest before produc…

8cbbd70

…ing demo records

docs(share-group): backtick ABI and 37.x to satisfy mdspell

bff1344

github-advanced-security AI found potential problems May 14, 2026

View reviewed changes

test(share-group): align assertions with Optional topic and non-null …

2d094d5

…OrderedPartitionableRecord data

FrankChen021 reviewed May 15, 2026

View reviewed changes

Conversation

Shekharrajak commented Apr 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Fixed the bug ...

Renamed the class ...

Added a forbidden-apis entry ...

Release note

Key changed/added classes in this PR

Demo

Uh oh!

Shekharrajak Apr 15, 2026

Choose a reason for hiding this comment

Uh oh!

Shekharrajak Apr 15, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Shekharrajak commented May 8, 2026

Uh oh!

Shekharrajak commented May 8, 2026

Uh oh!

Shekharrajak May 8, 2026

Choose a reason for hiding this comment

Uh oh!

Shekharrajak May 8, 2026

Choose a reason for hiding this comment

Uh oh!

Shekharrajak commented May 8, 2026

Uh oh!

Shekharrajak commented May 8, 2026

Uh oh!

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

FrankChen021 May 9, 2026

Choose a reason for hiding this comment

Uh oh!

Shekharrajak May 9, 2026

Choose a reason for hiding this comment

Uh oh!

Shekharrajak commented May 9, 2026

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Shekharrajak commented May 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Shekharrajak commented May 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Shekharrajak commented Apr 14, 2026 •

edited

Loading

Shekharrajak commented May 15, 2026 •

edited

Loading