KAFKA-19397: Not relaying on metadata to map between topic id and name. #19964

OmniaGM · 2025-06-13T14:29:34Z

Metadata doesn't have the full view of topicNames to ids during
rebootstrap of client or when topic has been deleted/recreated. The
solution is to pass down topic id and stop trying to figure it out later
in the logic.

- Metadata doesn't have the full view of topicNames to ids during rebootstrap of client or when topic has been deleted/recreated. The solution is to pass down topic id and stop trying to figure it out later in the logic.

junrao

@OmniaGM : Thanks for the PR. Overall, it looks good. A few comments below. Also, could we add a unit test?

@lucasbru : Could you test this PR with the stream job?

clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

junrao · 2025-06-14T06:44:47Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

+
+                    if (matchedBatchesForTopicId.size() > 1) {
+                        matchedBatchesForTopicId.forEach(matchedBatch ->
+                                failBatch(matchedBatch, new RuntimeException("More than one batch with same topic id and partition."), false));


If this is unexpected, we want to throw IllegalStateException as in other places in this file

Removed this now in favour of using first find. I am just thinking should we in general return IllegalStateException if batch is null?

lucasbru · 2025-06-16T09:45:54Z

@junrao I redeployed the soak with the fix. I will report back if the problem reoccurs

lucasbru · 2025-06-17T08:39:36Z

@OmniaGM @junrao I have been running the soak for 24h, and it's looking good.

chia7712

@OmniaGM thanks for this patch. a couple of comments are left. PTAL

chia7712 · 2025-06-17T13:51:12Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

+                    // topic name in the response might be empty.
+                    ProducerBatch batch = batches.entrySet().stream()
+                            .filter(entry ->
+                                    entry.getKey().same(new TopicIdPartition(r.topicId(), p.index(), r.name()))


Should we create the TopicIdPartition outside the stream to avoid creating many temporary objects?

moved the initalisation out of stream

chia7712 · 2025-06-17T13:52:31Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

+                    ProducerBatch batch = batches.entrySet().stream()
+                            .filter(entry ->
+                                    entry.getKey().same(new TopicIdPartition(r.topicId(), p.index(), r.name()))
+                            ).map(Map.Entry::getValue).findFirst().orElse(null);


It is possible to have null batch, right? For example, the topic is recreated after the batch is generated

We always had this potentail of batch is null this why I raised the comment here #19964 (comment) that should we have IllegalStateException. I updated this to fail with IllegalStateException instead of leaving it like this

clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java

chia7712 · 2025-06-17T14:22:34Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

@@ -855,7 +860,7 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
        if (batches.isEmpty())
            return;

-        final Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());
+        final Map<TopicIdPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());
        Map<String, Uuid> topicIds = topicIdsForBatches(batches);


one question: what happens if the topic id is unknown (Uuid.ZERO_UUID)?

We always fall back on topic name if topic id not passed down in the produce request and sort it out down the line when the broker recieve the produce request. It should be the same as if the client not topic id aware.

We always fall back on topic name if topic id not passed down

IIRC, the topic name is ignored in RPC when the version is larger than 13. So, is it possible that the broker receives the produce request without both the topic name and id?

It shouldn't happened but if it did the batch would fail with "unkonw topic"

The sender on client side always fetch metadata before send so we should get some sort of metadata to populate topic id

Actually there is a potential for this in Sender::topicIdsForBatches default to Uuid zero if metadata doesn't have the topic. Am thinking now that maybe ProducerBatch should use TopicIdPartition instead of TopicPartition

It's true that Sender::topicIdsForBatches could return zero topicId in rare cases. It seems that we could just pass along the mapping to zero topicId when handling the produce response.

OmniaGM · 2025-06-17T16:16:47Z

I have added some test in ProducerSendWhileDeletionTest to cover recreation while producing as well hope this will be enough to cover these cases.

kirktrue

Thanks for the PR @OmniaGM!

A few very minor comments.

Thanks!

clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java

….java Co-authored-by: Kirk True <[email protected]>

lucasbru

Needs a rebase, but LGTM

junrao

@OmniaGM : Thanks for the updated PR. A few more comments.

junrao · 2025-06-19T07:22:39Z

clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java

@@ -78,6 +80,21 @@ public TopicPartition topicPartition() {
        return topicPartition;
    }

+    /**
+     * Checking if TopicIdPartition meant to be the same reference to same this object but doesn't have all the data.


same this object => this object

junrao · 2025-06-19T07:38:28Z

clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java

+            return topicId.equals(tpId.topicId) &&
+                    topicPartition.partition() == tpId.partition();
+        } else {
+            return topicPartition.equals(tpId.topicPartition());


In the rare case that Sender::topicIdsForBatches returns 0 topic id (e.g. topic is deleted), we will pass along topicName -> 0 to handleProduceResponse(). The response will include empty topic and 0 topic id. It's important that we find a match in this case to avoid IllegalStateException. I am thinking that we should first try to do the comparison on topic name, if it's not empty. Otherwise, just do the comparison on topic id even if it's zero.

junrao · 2025-06-19T07:47:25Z

core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala

+    }
+  }
+
+  def recreateTopic(admin: Admin, topic: String): Unit = {


Could this be private?

junrao · 2025-06-19T07:50:57Z

core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala

+      fs.join()
+      producerFutures.forEach(_.join)
+    } catch {
+      case e: TopicExistsException =>


Hmm, where would a TopicExistsException be thrown?

junrao · 2025-06-19T07:56:09Z

core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala

+            producer.send(new ProducerRecord(topic, null, ("value" + i).getBytes(StandardCharsets.UTF_8)), new Callback() {
+              override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
+                assertNotNull(metadata)
+                assertNotEquals(metadata.offset, -1L)


Since the topic could be deleted, it's possible for the metadata to have -1, right?

Also, could we verify that every record's callback is called?

junrao · 2025-06-19T07:58:04Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

+
+                    if (batch == null) {
+                        throw new IllegalStateException("batch created for " + tpId + " can't be found, " +
+                                "topic might be recreated after the batch creation.");


Hmm, the recreation of the topic shouldn't hit this IllegalStateException, right?

junrao · 2025-06-19T08:01:31Z

core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala

+        topicMetadata(admin, topic).topicId()
+      } else Uuid.ZERO_UUID
+      // don't wait for the physical delete
+      deleteTopicWithAdminRaw(admin, topic)


Hmm, deleteTopicWithAdminRaw() doesn't wait for the metadata propagation to the brokers. However, the producer only sees the deleted topic after the metadata is propagated. Is this test effective?

junrao · 2025-06-19T08:03:18Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

+                    // We need to find batch based on topic id and partition index only as
+                    // topic name in the response might be empty.
+                    TopicIdPartition tpId = new TopicIdPartition(r.topicId(), p.index(), r.name());
+                    ProducerBatch batch = batches.entrySet().stream()


This changes a map lookup to an iteration. Could we do some produce perf test (with multiple topic/partitions) to verify there is no performance degradation?

github-actions bot added triage PRs from the community producer clients small Small PRs labels Jun 13, 2025

OmniaGM force-pushed the KAFKA-19397 branch from b90445d to 75129f7 Compare June 13, 2025 14:37

OmniaGM mentioned this pull request Jun 13, 2025

KAFKA-10551: Add topic id support to produce request and response #15968

Merged

chia7712 self-requested a review June 13, 2025 16:21

chia7712 added the ci-approved label Jun 13, 2025

OmniaGM force-pushed the KAFKA-19397 branch from 75129f7 to 324230c Compare June 13, 2025 17:29

junrao reviewed Jun 14, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Jun 15, 2025

Address feedback

3f948eb

chia7712 reviewed Jun 17, 2025

View reviewed changes

Add test cases to keep recreate topic while producing

bb6d82b

github-actions bot added core Kafka Broker and removed small Small PRs labels Jun 17, 2025

address chia's feedback

584740a

OmniaGM force-pushed the KAFKA-19397 branch from 5fdd621 to 584740a Compare June 17, 2025 16:10

throw IllegalStateException instead of failBatch when batch is null

b57e771

kirktrue reviewed Jun 17, 2025

View reviewed changes

clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java Outdated Show resolved Hide resolved

clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java Outdated Show resolved Hide resolved

OmniaGM and others added 2 commits June 18, 2025 12:50

Update clients/src/main/java/org/apache/kafka/common/TopicIdPartition…

1faf543

….java Co-authored-by: Kirk True <[email protected]>

address feedback

f47ed06

lucasbru approved these changes Jun 18, 2025

View reviewed changes

junrao reviewed Jun 19, 2025

View reviewed changes

KAFKA-19397: Not relaying on metadata to map between topic id and name. #19964

Are you sure you want to change the base?

KAFKA-19397: Not relaying on metadata to map between topic id and name. #19964

Uh oh!

Conversation

OmniaGM commented Jun 13, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lucasbru commented Jun 16, 2025

Uh oh!

lucasbru commented Jun 17, 2025

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

OmniaGM Jun 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

OmniaGM commented Jun 17, 2025

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

OmniaGM commented Jun 13, 2025 •

edited by github-actions bot

Loading

OmniaGM Jun 17, 2025 •

edited

Loading