KAFKA-10545: Create topic IDs and propagate to brokers by jolshan · Pull Request #9626 · apache/kafka

jolshan · 2020-11-19T23:48:46Z

This change takes the topic IDs created in #9473 and propagates them to brokers using LeaderAndIsr Request. It also removes the topic name from the LeaderAndIsr Response, reorganizes the response to be sorted by topic, and includes the topic ID.

In addition, the topic ID is persisted to each replica in Log as well as in a file on disk. This file is read on startup and if the topic ID exists, it will be reloaded.

This PR bumps the IBP and is expected to be merged at the same time as #9622 as to not bump the protocol twice

Committer Checklist (excluded from commit message)

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

dengziming

ApiVersionTest need to be changed because you changed ApiVersion, and a few suggesions.

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

dengziming · 2020-12-04T13:06:06Z

KAFKA-10729 add KAFKA_2_8_IV0, so you should bump the version again.

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

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

rite2nikhil

thanks for the pr, left a few commrnts

dajac

Thanks for the PR. I have left some comments. I have only looked at the request/response and the controller so far. I will look at the remaining parts tomorrow.

clients/src/main/resources/common/message/LeaderAndIsrRequest.json

clients/src/main/resources/common/message/LeaderAndIsrResponse.json

core/src/test/scala/unit/kafka/controller/ControllerChannelManagerTest.scala

dajac · 2020-12-10T15:42:52Z

core/src/main/scala/kafka/controller/KafkaController.scala

-        offlineReplicas += tp
-      else if (partition.errorCode == Errors.NONE.code)
-        onlineReplicas += tp
+    if (leaderAndIsrResponse.topics().isEmpty) {


It may be better to also have an explicit handling of the version here. Alternatively, we could push this into the LeaderAndIsrResponse and provides a method Map<TopicPartition, ...> partitions() which handles the version.

Actually, this may not work as we don't have the topic name in the latest version...

Yeah. One option I thought of would be to do something like partitions(Map<id, string> topicNames) and handle it inside the response with the version. That might be a little cleaner but I'm not sure.

core/src/main/scala/kafka/controller/KafkaController.scala

rajinisivaram

@jolshan Thanks for the PR, looks good. Left some minor comments.

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

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

rajinisivaram · 2020-12-15T11:14:59Z

core/src/main/scala/kafka/controller/KafkaController.scala

+    }
+
+    leaderAndIsrResponse.topics.forEach { topic =>
+      val topicName = controllerContext.topicNames.get(topic.topicId).get


Should we do controllerContext.topicNames.get(topic.topicId).foreach instead of get to avoid throwing exception if topic is not in the context?

I moved this logic to LeaderAndIsrResponse which is java code, so it behaves a bit differently. (We may want to move back though, and if so, I'll do this.) For now I do a null check in the java code.

One question I have though is what does it mean to not have a topic in the context? Is it possible to send a leaderAndIsrRequest with an unknown partition/one not in the context? I can only think maybe the topic gets deleted while handling the request. What would the normal error handling case be in that scenario?

core/src/main/scala/kafka/controller/KafkaController.scala

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

rajinisivaram · 2020-12-15T11:42:04Z

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

+                    log.partitionMetadataFile.get.write(topicIds.get(topicPartition.topic))
+                    log.topicId = topicIds.get(topicPartition.topic)
+                  } else {
+                    stateChangeLogger.warn("Partition metadata file already contains content.")


why is this a warning?

I think that if we reach here, we are in an unexpected state. The partitionMetadata file should have loaded in the topic ID if the file already contained content. I left as a warn rather than an exception.

Hmm, looking at the conditional statements here, it looks like we would write the file the first time we get here because log.partitionMetadataFile.get.isEmpty() and the second time we would print a warning even if the id in the file matches the expected id. Unless I missed something.

Oops. I think I cleaned up this block and deleted something. There should be a check if log.topicId.equals(id). If so, then the file exists and we shouldn't go in to the block that says "// There is not yet a topic ID stored in the log."

I should also fix the topicIds.get(topicPartition.topic) above and replace with id.

I might think about making this code cleaner in general to avoid so many nested if statements

So I thought through these cases some more and realized that the metadata file will fail to open if formatted incorrectly. So the only case where there could be data written to the file is if the ID is the zero UUID. So I decided to just fail on reading the file if the zero ID is provided. (We will never write zero ID to file.) The rest of this cleaned up pretty nicely.

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

rajinisivaram

@jolshan Thanks for the updates, looks good, just one comment/question left. The changes to process errors in LeaderAndIsrResponse.java seem fine. As you said, the only case where topic is not in the controller context would be where a topic was deleted. Don't think we need any other special handling for that. The PR needs rebasing to resolve conflicts btw.

rajinisivaram · 2020-12-16T10:59:33Z

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

+                    log.partitionMetadataFile.get.write(topicIds.get(topicPartition.topic))
+                    log.topicId = topicIds.get(topicPartition.topic)
+                  } else {
+                    stateChangeLogger.warn("Partition metadata file already contains content.")


Hmm, looking at the conditional statements here, it looks like we would write the file the first time we get here because log.partitionMetadataFile.get.isEmpty() and the second time we would print a warning even if the id in the file matches the expected id. Unless I missed something.

jolshan · 2020-12-16T16:30:20Z

@rajinisivaram Yup, will look at the rebase next.

rajinisivaram

@jolshan Thanks for the updates. Left a minor comment, apart from that LGTM. PR #9622 is also ready to merge, so we can merge these two together later today.

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

rajinisivaram · 2020-12-18T22:15:35Z

@jolshan Thanks for the PR, the last build was good, so merging to trunk

jolshan mentioned this pull request Nov 19, 2020

KAFKA-10547; add topicId in MetadataResp #9622

Merged

3 tasks

dengziming reviewed Nov 20, 2020

View reviewed changes

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

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

jolshan added 9 commits November 20, 2020 08:39

First implementation

ef2c75b

Merge branch 'trunk' of github.com:apache/kafka into KIP516

0e0a304

Merge branch 'trunk' of github.com:apache/kafka into KIP516

12e440b

Merge branch 'trunk' of github.com:apache/kafka into KIP516

b6b0796

Merge branch 'trunk' of github.com:apache/kafka into KIP516

0279e2c

Merge branch 'trunk' of github.com:apache/kafka into KIP516

0f0e327

Merge branch 'trunk' of github.com:apache/kafka into KIP516LeaderAndIsr

7fa1e33

Fixes

37cdd62

Cleaned up Uuid, merge processes

1926bf4

jolshan force-pushed the KIP516LeaderAndIsr branch from 5999b00 to 1926bf4 Compare November 20, 2020 19:09

rite2nikhil reviewed Dec 10, 2020

View reviewed changes

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

rite2nikhil reviewed Dec 10, 2020

View reviewed changes

clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java Show resolved Hide resolved

rite2nikhil reviewed Dec 10, 2020

View reviewed changes

dajac reviewed Dec 10, 2020

View reviewed changes

rajinisivaram reviewed Dec 15, 2020

View reviewed changes

Addressed comments

5424a1a

rajinisivaram reviewed Dec 16, 2020

View reviewed changes

jolshan added 2 commits December 16, 2020 17:20

Made replicaManager code less complicated.

6dbdbb8

Merge branch 'trunk' of github.com:apache/kafka into KIP516LeaderAndIsr

53ea843

rajinisivaram reviewed Dec 17, 2020

View reviewed changes

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala Outdated Show resolved Hide resolved

jolshan added 4 commits December 17, 2020 15:00

Fix some issues with propagating partition state

919f0ed

Remove extra code

fcd8093

Merge branch 'trunk' into KIP516LeaderAndIsr

213143d

Merge branch 'trunk' into KIP516LeaderAndIsr

5ff7840

rajinisivaram merged commit 1dd1e7f into apache:trunk Dec 18, 2020

chia7712 mentioned this pull request Dec 19, 2020

HOTFIX: fix failed ControllerChannelManagerTest#testUpdateMetadataReq… #9768

Merged

3 tasks

Comments

Conversation

jolshan commented Nov 19, 2020

Committer Checklist (excluded from commit message)

Uh oh!

dengziming left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

dengziming commented Dec 4, 2020

Uh oh!

Uh oh!

Uh oh!

rite2nikhil left a comment

Choose a reason for hiding this comment

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jolshan Dec 10, 2020 • 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!

rajinisivaram left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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!

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!

Uh oh!

Uh oh!

rajinisivaram left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jolshan commented Dec 16, 2020

Uh oh!

rajinisivaram left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

rajinisivaram commented Dec 18, 2020

Uh oh!

Reviewers

Assignees

Labels

jolshan Dec 10, 2020 •

edited

Loading