KAFKA-15585: Add DescribeTopics API server side support by CalvinLiu7947 · Pull Request #14612 · apache/kafka

CalvinLiu7947 · 2023-10-23T05:49:33Z

Introduce the DescribeTopics API and the server-side handling code.
https://issues.apache.org/jira/browse/KAFKA-15585

CalvinLiu7947 · 2023-11-18T19:03:02Z

Updated the API schema with the Cursor. It is needed both in request and response.
Removed the RequestLimitReached error.
Use ordered map in the TopicsImage.

mumrah

Thanks @CalvinConfluent!

Left some comments inline

mumrah · 2023-11-20T16:27:09Z

@CalvinConfluent btw, have you updated the KIP to reflect the two RPC schemas you've corrected here?

CalvinLiu7947 · 2023-11-20T17:26:14Z

@mumrah Thanks for the review, KIP updated.

artemlivshits · 2023-11-21T01:45:15Z

+    val cursor = describeTopicPartitionsRequest.cursor()
+    val fetchAllTopics = topics.isEmpty
+    if (fetchAllTopics) {
+      kRaftMetadataCache.getAllTopics().foreach(topic => topics.append(topic))


If we copy and sort all the topic names anyway, do we need to change the underlying data structure to NavigableMap? We could just use this list to traverse topic info and it will be in order.

In the fetch all path, no additional sort is required. I did not see a good way to convert Java list to a scala mutable list, so I did the copy.
Use a mutable list for 2 reasons

It is easier to filter out the topics alphabetically ahead of the cursor topic

In the fetch all case, I think we should still include the cursor topic in the response if it does not exist. Mutable list make it easier.

But if you ask whether it is worth the effort to create the full set of underline structures to get an ordered list where we can just sort the topic list, I am not sure.

CalvinLiu7947 · 2023-11-21T23:17:44Z

As discussed offline, we will focus on the pagination behavior.
The performance optimization is tracked in https://issues.apache.org/jira/browse/KAFKA-15873

mumrah

Thanks for the updates @CalvinConfluent! Looks like there are some conflicts with trunk.

I think we should add an integration "request" test for the new RPC. See ApiVersionsRequestTest for a basic example. We can also do this as a follow-up.

CalvinLiu7947 · 2023-11-28T21:09:31Z

@mumrah Thanks for the review. The integration tests will be introduced in the client side change.

This reverts commit b7cd9f2.

artemlivshits · 2024-01-12T06:36:52Z

+        val result = new ListBuffer[DescribeTopicPartitionsResponsePartition]()
+        val endIndex = upperIndex.min(topic.partitions().size())
+        for (partitionId <- startIndex until endIndex) {
+          val partition = topic.partitions().get(partitionId)


What if partition doesn't exist?

Do you mean the partitions in the topic are not consecutive? Just realize it is possible.

Actually it is not possible, the partition index starts with 0 and increments by 1.
Then what is the case if the partition does not exist?

The data structure leaves a possibility (due to a bug or a change elsewhere) to have arbitrary numbers. It would be good not to crash if the current assumptions are violated.

Sure, updated.

artemlivshits · 2024-01-12T06:51:33Z

+        if (!partitionResponse.isDefined) {
+          val error = try {
+            Topic.validate(topicName)
+            Errors.UNKNOWN_TOPIC_OR_PARTITION


Yeah, but the error is kind of unexpected -- if the user didn't specify a topic in the first place, why would it get an error about a topic that doesn't exist?

mumrah

Thanks for the updates @CalvinConfluent, I like the new iterator approach. I left just one comment on that inline.

I also like that you wrote the new request handler in Java. I think that's a first 😄

artemlivshits · 2024-01-12T22:43:26Z

-                .setEligibleLeaderReplicas(Replicas.toList(partition.elr))
-                .setLastKnownElr(Replicas.toList(partition.lastKnownElr)))
+        // The partition id may not be consecutive.
+        val partitions = topic.partitions().keySet().stream().sorted().iterator()


This has O(N*logN) runtime complexity and O(N) space complexity. We could do O(N) complexity and not have an extra copy if we just iterate over all partitions and filter the ones that fit into the required range (one of your previous implementations had this).

I am not sure I get it. The partition IDs can be random like the cases in UT, I don't have an O(n) with no extra space simple solution off the top of my head. Maybe running the quick select can do the trick but it is not generically supported by Java.
Instead, I use a tree set to maintain the top K smallest partitions larger than the start index. This is better than the original sorting.

artemlivshits · 2024-01-17T22:01:09Z

-                  .setEligibleLeaderReplicas(Replicas.toList(partition.elr))
-                  .setLastKnownElr(Replicas.toList(partition.lastKnownElr)))
-            }
+        val partitions = topic.partitions().keySet()


Looks like here we just need to remember the size? Or maybe calculate the nextIndex directly here?

artemlivshits · 2024-01-18T03:07:25Z

+        val result = new ListBuffer[DescribeTopicPartitionsResponsePartition]()
+        val endIndex = upperIndex.min(topic.partitions().size())
+        for (partitionId <- startIndex until endIndex) {
+          val partition = topic.partitions().get(partitionId)


The data structure leaves a possibility (due to a bug or a change elsewhere) to have arbitrary numbers. It would be good not to crash if the current assumptions are violated.

artemlivshits · 2024-01-18T03:13:33Z

+          val maybeLeader = getAliveEndpoint(image, partition.leader, listenerName)
+          maybeLeader match {
+            case None =>
+              val error = if (!image.cluster().brokers.containsKey(partition.leader)) {


I guess we need to see what the client does with the error code.

CalvinLiu7947 · 2024-01-19T07:36:23Z

Verified the following tests locally
testDescribeUnderReplicatedPartitionsWhenReassignmentIsInProgress also fails in other PR https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2588/tests/
testDescribeQuorumReplicationSuccessful
This PR mostly new code and uses its code path, so theoretically will not affect other UT. Running the integration again by merging the latest master.

mumrah

Thanks for all the work on this @CalvinConfluent. LGTM

Please double check that the failing tests on Jenkins look okay locally.

CalvinLiu7947 · 2024-01-24T18:09:54Z

@mumrah Thanks! I have verified the tests failing can pass locally.

…14612) This patch implements the new DescribeTopicPartitions RPC as defined in KIP-966 (ELR). Additionally, this patch adds a broker config "max.request.partition.size.limit" which limits the number of partitions returned by the new RPC. Reviewers: Artem Livshits <alivshits@confluent.io>, Jason Gustafson <jason@confluent.io>, David Arthur <mumrah@gmail.com>

…mnative#942) Main changes: - Adapt to the new `AddPartitionsToTxnRequest` from apache/kafka#13231 (KIP-890) - Support the new `DescribeTopicPartitions` request from apache/kafka#14612 (KIP-966), which is required by some admin APIs Other changes: - apache/kafka#13760 will retry when `deleteRecords` returns a retriable error, change the error code to `INVALID_REQUEST`

CalvinLiu7947 commented Oct 31, 2023

View reviewed changes

Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated

hachikuji reviewed Nov 15, 2023

View reviewed changes

CalvinLiu7947 force-pushed the ELR-ak-Describe-topics-api branch from 637d45a to 35f9763 Compare November 15, 2023 05:35

mumrah reviewed Nov 20, 2023

View reviewed changes

artemlivshits reviewed Nov 21, 2023

View reviewed changes

CalvinLiu7947 requested review from artemlivshits, hachikuji and mumrah November 21, 2023 15:57

mumrah reviewed Nov 28, 2023

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/common/requests/DescribeTopicPartitionsResponse.java Outdated

CalvinLiu7947 force-pushed the ELR-ak-Describe-topics-api branch from aa1c51b to b2bdf53 Compare November 28, 2023 21:07

CalvinLiu7947 added 15 commits January 3, 2024 10:29

Add DescribeTopics API server side support

9793aa1

Minor

c201fca

Address comment

9ea171d

Minor

3664054

Update the schema.

19b13be

Remove RequestLimitReached

38ee839

Use ordered map for TopicsImage

c8c5d62

Minor

dadc3a3

Update the schema to include the partition limit.

8f9ffd4

Minor

5926b45

Address comments

943c116

Revert "Use ordered map for TopicsImage"

2683669

This reverts commit b7cd9f2.

Address comment

7abb135

Fix UT

d21eda7

Fix UT

f2d5990

CalvinLiu7947 requested a review from artemlivshits January 11, 2024 05:16

artemlivshits reviewed Jan 12, 2024

View reviewed changes

CalvinLiu7947 added 2 commits January 12, 2024 11:42

Address comments

0559cf0

Minor

93708e2

mumrah reviewed Jan 12, 2024

View reviewed changes

Comment thread core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java Outdated

Comment thread core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java Outdated

artemlivshits reviewed Jan 12, 2024

View reviewed changes

Address comments

dfcaf66

CalvinLiu7947 requested review from artemlivshits and mumrah January 13, 2024 00:08

CalvinLiu7947 added 3 commits January 16, 2024 15:09

Using index range to check partitions

7ee72aa

Merge branch 'trunk' into ELR-ak-Describe-topics-api

8a30c74

Minor

f577396

artemlivshits reviewed Jan 18, 2024

View reviewed changes

CalvinLiu7947 added 2 commits January 17, 2024 20:49

Address comments

dcad940

Merge branch 'trunk' into ELR-ak-Describe-topics-api

7489ac7

CalvinLiu7947 requested a review from artemlivshits January 18, 2024 16:28

Minor

f8eb307

mumrah reviewed Jan 18, 2024

View reviewed changes

Comment thread core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala Outdated

Address comment

9fb9ce3

CalvinLiu7947 requested a review from mumrah January 18, 2024 22:40

CalvinLiu7947 added 2 commits January 18, 2024 23:37

Merge branch 'trunk' into ELR-ak-Describe-topics-api

79d04c2

Merge branch 'trunk' into ELR-ak-Describe-topics-api

7f32ea4

mumrah approved these changes Jan 24, 2024

View reviewed changes

mumrah merged commit 7e5ef9b into apache:trunk Jan 24, 2024

Conversation

CalvinLiu7947 commented Oct 23, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

CalvinLiu7947 commented Nov 18, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mumrah left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mumrah commented Nov 20, 2023

Uh oh!

CalvinLiu7947 commented Nov 20, 2023

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

CalvinLiu7947 Nov 21, 2023 • 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!

CalvinLiu7947 commented Nov 21, 2023

Uh oh!

mumrah left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

CalvinLiu7947 commented Nov 28, 2023

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!

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!

mumrah 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!

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!

CalvinLiu7947 commented Oct 23, 2023 •

edited

Loading

CalvinLiu7947 commented Nov 18, 2023 •

edited

Loading

CalvinLiu7947 Nov 21, 2023 •

edited

Loading