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

OmniaGM · 2024-05-15T15:53:10Z

Add support topicId in ProduceRequest/ProduceResponse. Topic name
and Topic Id will become ignorable following the footstep of
FetchRequest/FetchResponse
ReplicaManager still look for HostedPartition using TopicPartition
and doesn't check topic id. This is an [OPEN QUESTION] if we should
address this in this pr or wait for
KAFKA-16212 as this
will update ReplicaManager::getPartition to use TopicIdParittion
once we update the cache. Other option is that we compare provided
topicId with Partition topic id and return UNKNOW_TOPIC_ID or
UNKNOW_TOPIC_PARTITION if we can't find partition with matched topic
id.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…equest

OmniaGM · 2024-05-16T13:01:33Z

Few of the failed tests are related to this change and am working on fixing them

OmniaGM · 2024-05-20T12:00:14Z

Few of the failed tests are related to this change and am working on fixing them

I believe that failed tests now are unrelated

jolshan · 2024-05-20T22:29:30Z

Topic name and Topic Id will become optional following the footstep of FetchRequest/FetchResponse

My understanding is that all requests going forward will use ID and not name similar to fetch request. I believe that is what is in the PR, but the comment suggests otherwise.

jolshan · 2024-05-20T22:41:11Z

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

@@ -610,7 +611,9 @@ private void handleProduceResponse(ClientResponse response, Map<TopicPartition,
                // This will be set by completeBatch.
                Map<TopicPartition, Metadata.LeaderIdAndEpoch> partitionsWithUpdatedLeaderInfo = new HashMap<>();
                produceResponse.data().responses().forEach(r -> r.partitionResponses().forEach(p -> {
-                    TopicPartition tp = new TopicPartition(r.name(), p.index());
+                    // Version 12 drop topic name and add support to topic id. However, metadata can be used to map topic id to topic name.
+                    String topicName = (r.name() == null || r.name().isEmpty()) ? metadata.topicNames().get(r.topicId()) : r.name();


What do we do if metadata has refreshed and is no longer in the metadata?
For fetch it is a bit different since we have the session logic, and can handle missing topics.

I would recommend writing through a few cases where the server and client have/don't have the topic ID to reason about the upgrade case/downgrade case/deletions/reassignments.

If topic has been recreated and topic id is out of date, the client will get UNKNOWN_TOPIC_ID and on the retry the topic id will be updated

If topic has been reassigned to another broker then the client will get NOT_LEADER_OR_FOLLOWER and then the client can retry with the right broker.

Am not sure what upgrade case/downgrade you refer too here Do you mean the client and broker IBP combination? If yes then some of these are covered in ProduceRequestTest and RequestResponseTest

I added two test cases to cover the first two and the producer seem to self recover on retry.

Yes. For the fetch request for example, there is code to make sure that all topics have IDs before we can send the fetch request. This is a bit less of an issue now, but if we have a cluster that is running on a MV < 2.8, topics will not have IDs. So when we decide which version of produce we want to send, we want to be aware of this.

Not only that, but even if the broker supports topic IDs on all topics, we also may have a case where we need to do a rolling upgrade to get the code that supports handling the latest API version. This may be less complicated for Produce since it is a client only API and doesn't rely on MV/IBP, so the apiVersions exchange between the client and the broker may be enough to ensure api compatibility.

We just want to confirm these upgrade paths are compatible since produce is the hot path and we don't want any (or at least not extended) downtime in the middle of an upgrade.

Hi @jolshan sorry for taking too long to get back to you on this. I have added ApiVersions::maxSupportedProduceVersion to determine what the supported version supported on the cluster. Regarding the tests for these combination of old/new producer with old/new broker are all covered in the system tests tests/kafkatest/tests/core/compatibility_test_new_broker_test.py and tests/kafkatest/tests/client/client_compatibility_produce_consume_test.py. I am trying to add another one to test the same while deleting topic and leadership changes but not sure if we really need to cover this or not as these are covered in the integration tests. WDYT?

No problem. I've also been super busy so not sure if I would have had time to review.

As long as the tests are between an older version and a newer version (ie not upgrade tests where the issue can just seem transient) then this should be ok.

Whenever I implemented the fetch change, I also did some sanity checks by looking at the logs when upgrading to see the change in request version without errors. I'd recommend doing this at least once as a manual test.

Thanks for the change and I will take a look.

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

core/src/main/scala/kafka/server/KafkaApis.scala

jolshan · 2024-05-20T23:10:47Z

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -1361,10 +1366,10 @@ class ReplicaManager(val config: KafkaConfig,
   */
  private def appendToLocalLog(internalTopicsAllowed: Boolean,
                               origin: AppendOrigin,
-                               entriesPerPartition: Map[TopicPartition, MemoryRecords],
+                               entriesPerPartition: Map[TopicIdPartition, MemoryRecords],


is there a reason to pass this data structure here if we are not using the ID to check the append at the log level?

two reasons here

I didn't want to keep convert between TopicIdPartitions to TopicPartition

KAFKA-16212 will eventually use TopicIdPartitions to getPartitionOrException

Ok -- once we start using these across the log layer it makes sense.

Did we end up doing the getPartitionOrException in this PR?

We introduced getPartitionOrException(topicIdPartition: TopicIdPartition) that verifies the topicId if provided.

jolshan · 2024-05-20T23:31:21Z

I would recommend taking a look at where we are passing the topic ID through and the checks we do. If we think it is useful to ensure we are writing to the right topic, we should do it, but if it is just adding complexity, we may want to consider changing.

OmniaGM · 2024-05-22T09:42:02Z

Topic name and Topic Id will become optional following the footstep of FetchRequest/FetchResponse

My understanding is that all requests going forward will use ID and not name similar to fetch request. I believe that is what is in the PR, but the comment suggests otherwise.

I meant that in Json files both will be marked ignorable

…equest

ijuma · 2024-12-22T21:19:45Z

Looks like this needs a rebase or merge from master.

OmniaGM · 2025-01-17T15:17:35Z

Looks like this needs a rebase or merge from master.

I will have a look into this next week

jolshan · 2025-02-06T21:17:03Z

@OmniaGM any updates here? Thanks for your patience and work so far!

OmniaGM · 2025-03-17T06:39:48Z

@OmniaGM any updates here? Thanks for your patience and work so far!

Hi, Sorry for this very late reply. I have been busy with too many things lately and haven't had time to look into resolving the conflicts. I'll try to work on this as soon as possible. Thanks for your patience

ijuma · 2025-03-21T16:59:51Z

@OmniaGM Would you like someone to help get this over the finish line?

…equest

OmniaGM · 2025-04-02T16:00:27Z

@OmniaGM Would you like someone to help get this over the finish line?

I am nearly done with fixing the conflict should publish this this week. sorry for late replay I have been away on holiday for the last 2 weeks

…equest

ijuma · 2025-04-02T16:55:16Z

That's great, thanks!

…equest

OmniaGM · 2025-04-04T10:44:48Z

That's great, thanks!

@ijuma and @jolshan I fixed the conflicts now

junrao

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

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

junrao · 2025-04-07T21:35:27Z

core/src/main/scala/kafka/server/KafkaApis.scala

+        val topicPartition = new TopicPartition(topicName, partition.index())
+        if (topicName.isEmpty)
+          nonExistingTopicResponses += new TopicIdPartition(topicId, topicPartition) -> new PartitionResponse(Errors.UNKNOWN_TOPIC_ID)
+        else if (!metadataCache.contains(topicPartition))


It seems that we don't need this since it's handled in ReplicaManager too.

The condition was there before my changes. I just moved it

core/src/main/scala/kafka/server/KafkaApis.scala

core/src/main/scala/kafka/server/ReplicaManager.scala

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

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

clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java

junrao

@OmniaGM : Thanks for the updated PR. A few more comments. Also, are the test failures related to this PR?

junrao · 2025-04-10T18:41:52Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

-    private final TopicPartition tp2 = new TopicPartition("test", 2);
+    private static final String TOPIC_NAME = "test";
+    private static final Uuid TOPIC_ID = Uuid.fromString("MKXx1fIkQy2J9jXHhK8m1w");
+    private static final Map<String, Uuid> TOPIC_IDS = new HashMap<>(Map.of(TOPIC_NAME, TOPIC_ID));


new HashMap<>(Map.of(TOPIC_NAME, TOPIC_ID)); => Map.of(TOPIC_NAME, TOPIC_ID)); ?

Map.of create immutable map which can't be edited later to add other topics needed for other test cases

junrao · 2025-04-10T18:55:12Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

    }

    @Test
    public void testTransactionalSplitBatchAndSend() throws Exception {
+        Uuid topicId = Uuid.fromString("2J9hK8m1wHMKjXfIkQyXx1");
+        TOPIC_IDS.put("testSplitBatchAndSend", topicId);


Hmm, topicId in this test is the same as the one in testIdempotentSplitBatchAndSend. Will this cause confusion during testing since TOPIC_IDS accumulates across tests?

They are the same topic ids and name. I am not sure if the topic name and id should move to the setup of the test suite as multiple test cases need this topic. WDYT?

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

clients/src/test/java/org/apache/kafka/common/requests/ProduceRequestTest.java

core/src/main/scala/kafka/server/ReplicaManager.scala

junrao · 2025-04-10T19:10:43Z

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

+
+      val topicId = topicNames.find { case (_, topicName) => topicName == topic}
+        .map { case (topicId, _) => topicId }
+        .getOrElse(Uuid.ZERO_UUID)


Hmm, should we only set topicId if version is <= 12? The produce response is only keyed on topic name when version is <= 12.

I am not sure I haver version at this point in the test to decide also, this is not to set topic id but rather to call find(topicName, topicId) which can't be called without topicId for old versions this would be Uuid.ZERO_UUID isn't?

It seems that this test always sends produce requests with the latest version. Perhaps we could add a comment and remove if (version >= 13) ?

junrao · 2025-04-10T20:12:16Z

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

@@ -33,8 +33,8 @@ import scala.jdk.CollectionConverters._

 class ProducerSendWhileDeletionTest extends IntegrationTestHarness {
  val producerCount: Int = 1
-  val brokerCount: Int = 2
-  val defaultLingerMs: Int = 5;
+  val brokerCount: Int = 3


Why does the tests need 3 brokers?

I think due to the reassignment test? There may be a way to do it with just 2 though.

as @jolshan mentioned this for easier test for reassignment

core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala

jolshan · 2025-04-10T23:01:15Z

clients/src/test/java/org/apache/kafka/common/message/MessageTest.java

@@ -886,7 +888,6 @@ public void testProduceResponseVersions() throws Exception {
        testAllMessageRoundTrips(new ProduceResponseData()
            .setResponses(new ProduceResponseData.TopicProduceResponseCollection(singletonList(
                new ProduceResponseData.TopicProduceResponse()
-                    .setName(topicName)


Did we remove this for version compatibility? I would think name or ID would be returned in the response?

Yes, topic name or id is added below based on the version.

junrao · 2025-04-11T17:13:28Z

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

+    }
+
+    val reassignment = Map(
+      partition0 -> Optional.of(new NewPartitionReassignment(util.Arrays.asList(1, 2))),


Hmm, the partition started with 1 replica, we should assign it to a single replica, right? We will need to adjust the comment below too.

junrao · 2025-04-11T17:16:08Z

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -1361,10 +1366,10 @@ class ReplicaManager(val config: KafkaConfig,
   */
  private def appendToLocalLog(internalTopicsAllowed: Boolean,
                               origin: AppendOrigin,
-                               entriesPerPartition: Map[TopicPartition, MemoryRecords],
+                               entriesPerPartition: Map[TopicIdPartition, MemoryRecords],


We introduced getPartitionOrException(topicIdPartition: TopicIdPartition) that verifies the topicId if provided.

…equest

OmniaGM added 4 commits May 8, 2024 14:46

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

d3acdf0

KAFKA-10551: fix compatibility with 2.6 IBP

ddeac1b

KAFKA-10551: refactor

a92abe6

Merge remote-tracking branch 'apache/trunk' into KAFKA-10551-produceR…

afede12

…equest

OmniaGM marked this pull request as draft May 15, 2024 15:53

fix test

a8f0c91

OmniaGM force-pushed the KAFKA-10551-produceRequest branch from 1805c4f to a8f0c91 Compare May 15, 2024 22:05

OmniaGM marked this pull request as ready for review May 15, 2024 22:58

OmniaGM changed the title ~~Kafka-10551: Add topic id support to produce request and response~~ KAFKA-10551: Add topic id support to produce request and response May 15, 2024

fix integration tests

8c3602b

OmniaGM force-pushed the KAFKA-10551-produceRequest branch from 63a6032 to 8c3602b Compare May 16, 2024 15:06

jolshan reviewed May 20, 2024

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java Outdated Show resolved Hide resolved

jolshan reviewed May 20, 2024

View reviewed changes

core/src/main/scala/kafka/server/KafkaApis.scala Outdated Show resolved Hide resolved

jolshan reviewed May 20, 2024

View reviewed changes

OmniaGM added 2 commits May 22, 2024 15:03

Merge branch 'trunk' into KAFKA-10551-produceRequest

cba0a2d

address part of the feedback

27ed97b

OmniaGM force-pushed the KAFKA-10551-produceRequest branch from 8daeb62 to 1abc2ac Compare May 28, 2024 15:43

Add testing for while recreate the topic and reassignment

35dba4b

OmniaGM force-pushed the KAFKA-10551-produceRequest branch from 1abc2ac to 35dba4b Compare May 28, 2024 16:02

Fetch supported produce version from apiVersions.

916e471

OmniaGM force-pushed the KAFKA-10551-produceRequest branch from 79814ff to 916e471 Compare June 11, 2024 21:22

Fetch supported produce version from apiVersions.

c59b10c

OmniaGM force-pushed the KAFKA-10551-produceRequest branch from 86c1f71 to c59b10c Compare June 11, 2024 21:25

Merge remote-tracking branch 'apache/trunk' into KAFKA-10551-produceR…

65d4a41

…equest

addressing some of @junrao feedback

900c460

AndrewJSchofield removed the KIP-932 Queues for Kafka label Jan 13, 2025

Merge remote-tracking branch 'apache/trunk' into KAFKA-10551-produceR…

b091e28

…equest

Merge remote-tracking branch 'apache/trunk' into KAFKA-10551-produceR…

9b6a892

…equest

fix merge

0ba7431

github-actions bot added connect mirror-maker-2 KIP-932 Queues for Kafka labels Apr 2, 2025

OmniaGM added 2 commits April 3, 2025 10:17

Merge remote-tracking branch 'apache/trunk' into KAFKA-10551-produceR…

628371a

…equest

Merge remote-tracking branch 'apache/trunk' into KAFKA-10551-produceR…

6cd5d4e

…equest

junrao added the ci-approved label Apr 7, 2025

junrao reviewed Apr 8, 2025

View reviewed changes

address feedback

2f37078

AndrewJSchofield removed the KIP-932 Queues for Kafka label Apr 10, 2025

junrao reviewed Apr 10, 2025

View reviewed changes

jolshan reviewed Apr 10, 2025

View reviewed changes

junrao reviewed Apr 11, 2025

View reviewed changes

OmniaGM added 2 commits April 16, 2025 09:02

address feedback

aa811f1

Merge remote-tracking branch 'apache/trunk' into KAFKA-10551-produceR…

f158714

…equest

github-actions bot added the KIP-932 Queues for Kafka label Apr 16, 2025

OmniaGM added 2 commits April 16, 2025 10:58

fix test

cb66c85

address feedback

6932bd1

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

Are you sure you want to change the base?

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

Conversation

OmniaGM commented May 15, 2024 • edited by github-actions bot Loading

Committer Checklist (excluded from commit message)

OmniaGM commented May 16, 2024

OmniaGM commented May 20, 2024

jolshan commented May 20, 2024 • edited Loading

Choose a reason for hiding this comment

OmniaGM May 28, 2024 • edited Loading

Choose a reason for hiding this comment

jolshan May 31, 2024 • edited Loading

Choose a reason for hiding this comment

OmniaGM Jun 11, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jolshan commented May 20, 2024

OmniaGM commented May 22, 2024

ijuma commented Dec 22, 2024 • edited Loading

OmniaGM commented Jan 17, 2025

jolshan commented Feb 6, 2025

OmniaGM commented Mar 17, 2025

ijuma commented Mar 21, 2025

OmniaGM commented Apr 2, 2025

ijuma commented Apr 2, 2025

OmniaGM commented Apr 4, 2025

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

OmniaGM commented May 15, 2024 •

edited by github-actions bot

Loading

jolshan commented May 20, 2024 •

edited

Loading

OmniaGM May 28, 2024 •

edited

Loading

jolshan May 31, 2024 •

edited

Loading

OmniaGM Jun 11, 2024 •

edited

Loading

ijuma commented Dec 22, 2024 •

edited

Loading