KAFKA-19441: encapsulate MetadataImage in GroupCoordinator/ShareCoordinator #20061

zzbennett · 2025-06-28T02:30:36Z

The MetadataImage has a lot of stuff in it and it gets passed around in many places in the new GroupCoordinator. This makes it difficult to understand what metadata the group coordinator actually relies on and makes it too easy to use metadata in ways it wasn't meant to be used.

This change encapsulate the MetadataImage in an interface (CoordinatorMetadataImage) that indicates and controls what metadata the group coordinator actually uses. Now it is much easier at a glance to see what dependencies the GroupCoordinator has on the metadata. Also, now we have a level of indirection that allows more flexibility in how the GroupCoordinator is provided the metadata it needs.

…inator

…9441

dajac

@zzbennett Thanks for the patch. I just made a pass and I left some high level comments. I am not entirely happy with the interface yet.

dajac · 2025-06-30T08:45:34Z

...mmon/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorMetadataImage.java

+    default Optional<TopicMetadata> topicMetadata(String topicName) {
+        var topicId = topicId(topicName);
+        return topicId.isEmpty() ? Optional.empty() : topicMetadata(topicId.get());
+    }


nit: Should we provide the default implementation for the topic id version in order to be consistent with partitionCount?

dajac · 2025-06-30T08:46:09Z

...mmon/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorMetadataImage.java

+
+    CoordinatorMetadataDelta emptyDelta();
+
+    Long version();


nit: Could we use long here?

dajac · 2025-06-30T08:48:01Z

...mmon/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorMetadataImage.java

+        default Set<Integer> partitionSet() {
+            return IntStream.range(0, partitionCount()).boxed().collect(Collectors.toSet());
+        }


This is annoying. Let me check where we still use this.

dajac · 2025-06-30T08:53:55Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java

@@ -1921,14 +1921,14 @@ private CompletableFuture<DeleteShareGroupOffsetsResponseData> handleDeleteShare
                errorTopicResponses.add(
                    new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic()
                        .setTopicId(topicData.topicId())
-                        .setTopicName(metadataImage.topics().topicIdToNameView().get(topicData.topicId()))
+                        .setTopicName(metadataImage.topicName(topicData.topicId()).orElse("<UNKNOWN>"))


We cannot return <UNKNOWN>. What was returned with the previous implementation? null?

Yea before we were returning null. I had seen some other places where we were setting topic name to <UNKNOWN> e.g. https://github.com/apache/kafka/blob/trunk/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java#L4924 but to maintain the existing behavior we can just keep it as null

dajac · 2025-06-30T08:54:18Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java

                        .setErrorMessage(Errors.forCode(errItem.get().errorCode()).message())
                        .setErrorCode(errItem.get().errorCode())
                );
            } else {
                successTopics.put(
                    topicData.topicId(),
-                    metadataImage.topics().topicIdToNameView().get(topicData.topicId())
+                    metadataImage.topicName(topicData.topicId()).orElse("<UNKNOWN>")


ditto. In this case, we can return ZERO_UUID.

this is setting the topic name here so I'm not sure if ZERO_UUID makes sense

dajac · 2025-06-30T09:01:31Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java

+            List<String> partitionRacks = new ArrayList<>(racks.get(i));
+            // topicMetadata returns an unmodifiable list
+            Collections.copy(partitionRacks, racks.get(i));


This is a tad annoying too because we already allocate a new ArrayList when partitionRacks is called. For performance reasons, I wonder if we should not return an unmodifiable list so we can reuse it here.

This part is performance sensitive. We have a jmh benchmark that we may want to run to ensure that we don't regress here.

Sure I can make it a modifiable list. Making it unmodifiable is probably overkill since, like you said, we already make a copy in the partitionRacks method. If we avoid the extra copy, then the new implementation should be equivalent to the old but I'll double check with the jmh benchmarks

I ran the group coordinator jmh benchmarks locally a couple times and didn't see any statistically significant variations in performance

dajac · 2025-06-30T09:03:20Z

...or/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java

+        }
+
+        CoordinatorMetadataImage.TopicMetadata topicMetadata = topicMetadataOp.get();
+        List<String> racks = topicMetadata.partitionRacks().get(partition);


This call is pretty nasty because it generate the entire map just to return the racks of a single partition and it does so every time racksForPartition is called. We should definitely consider my suggestion for the interface.

yea agreed...

dajac · 2025-06-30T09:03:38Z

...or/src/main/java/org/apache/kafka/coordinator/group/modern/SubscribedTopicDescriberImpl.java

+        CoordinatorMetadataImage.TopicMetadata topicMetadata = topicMetadataOp.get();
+        List<String> racks = topicMetadata.partitionRacks().get(partition);
+        if (racks == null) {
+            return Collections.emptySet();


nit: Set.of().

dajac · 2025-06-30T09:06:37Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

@@ -477,7 +482,7 @@ public void testMemberJoinsEmptyConsumerGroup() {
        MockPartitionAssignor assignor = new MockPartitionAssignor("range");
        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
            .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
-            .withMetadataImage(metadataImage)
+            .withMetadataImage(new KRaftCoordinatorMetadataImage(metadataImage))


I am debating whether we should update MetadataImageBuilder to return a CoordinatorMetadataImage instead of wrapping it everywhere. Have you considered it?

I actually did at some point add a method to the MetadataImageBuilder to return a CoordinatorMetadataImage but I must have given up updating the test to use the new method in all the places where it could. This test is really massive 😅

dajac · 2025-06-30T09:10:24Z

...mmon/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorMetadataImage.java

+
+    Optional<TopicMetadata> topicMetadata(Uuid topicId);
+
+    boolean shareGroupsEnabled();


This feels weird in the interface. I wonder whether we could expose the features instead. It seems useful for other needs too.

Yea it is weird. The share group coordinator relies on it in its onNewMetadataImage method https://github.com/apache/kafka/blob/trunk/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java#L1102

I just realized though that this particular instance of onNewMetadataImage is not part of any of the other coordinator related interfaces though, so I can easily just add the FeatureImage as an extra arg to the ShareCoordinator interface instead of handling it through the CoordinatorMetadataImage.

zzbennett · 2025-06-30T11:48:33Z

@dajac thanks for your review. I think I've addressed all your comments in e66548e

dajac · 2025-06-30T13:45:21Z

...mmon/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorMetadataImage.java

+
+        int partitionCount();
+
+        default Set<Integer> partitionSet() {


Could we remove it?

sure, I can move it out of the interface to keep the interface more clean. There are only 4 places where we use it so I can move the logic inline in those areas

lbradstreet · 2025-07-01T06:04:32Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java

@@ -1921,14 +1921,14 @@ private CompletableFuture<DeleteShareGroupOffsetsResponseData> handleDeleteShare
                errorTopicResponses.add(
                    new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic()
                        .setTopicId(topicData.topicId())
-                        .setTopicName(metadataImage.topics().topicIdToNameView().get(topicData.topicId()))
+                        .setTopicName(metadataImage.topicName(topicData.topicId()).orElse(null))


Should you ever get null here? I do see that it matches the existing behavior.

It probably wouldn't ever happen but returning null at least matches the current behavior and avoids throwing an exception if the Optional is empty

lbradstreet · 2025-07-01T06:05:14Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java


    /**
     * The cache for topic hash value by topic name.
     * A topic hash is calculated when there is a group subscribes to it.
-     * A topic hash is removed when it's updated in MetadataImage or there is no group subscribes to it.
+     * A topic hash is removed when it's updated in GroupCoordinatorMetadataImage or there is no group subscribes to it.


Should this be CoordinatorMetadataImage?

A yea good catch

Found one other place where it was GroupCoordinatorMetadataImage. At first that was what the class was called, but I renamed it to CoordinatorMetadataImage when I realized it was being shared by the Share Group Coordinator too

lbradstreet · 2025-07-01T06:12:31Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

-                        .collect(Collectors.toSet()),
-                    currentTimeMs
-                ));
+            metadataImage.topicMetadata(topic.topicName()).ifPresentOrElse(


This isn't really the ifPresentOrElse's fault but I do find these if blocks a little unwieldy. I don't really have a great suggestion right now -- it's probably better that we leave the current structure so that the refactoring is still fairly mechanical.

By current structure, you mean the if/else block instead of the ifPresenetOrElse? I'm fine with either one. I agree the ifPresentOrElse is kind of hard to read when it's a big block of logic. It's probably better for one or two line pieces of logic

Updated it to be a regular if/else block

…9441

zzbennett · 2025-07-03T07:19:31Z

...mmon/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorMetadataDelta.java

+
+    CoordinatorMetadataDelta EMPTY = emptyDelta();
+
+    interface TopicsDelta {


now I'm thinking of removing TopicsDelta and moving createdTopicsIds/changedTopicIds/deletedTopicsIds to the top level interface because it's kind of redundant to have this sub-class here.

zzbennett added 2 commits June 28, 2025 09:52

KAFKA-19441: encapsulate MetadataImage in GroupCoordinator/ShareCoord…

c582944

…inator

minor cleanup

0fefdb3

github-actions bot added triage PRs from the community core Kafka Broker tools performance KIP-932 Queues for Kafka group-coordinator labels Jun 28, 2025

TaiJuWu added the ci-approved label Jun 29, 2025

zzbennett added 3 commits June 30, 2025 17:11

fix linting

d089ed0

Merge branch 'trunk' of github.com:apache/kafka into lbennett-kafka-1…

12aa519

…9441

fix linting

12d593b

dajac reviewed Jun 30, 2025

View reviewed changes

PR feedback

e66548e

github-actions bot added the kraft label Jun 30, 2025

dajac reviewed Jun 30, 2025

View reviewed changes

zzbennett added 2 commits July 1, 2025 08:07

remove the partitionSet() function from TopicMetadata interface

5bd417f

fix linting

adc7ce1

github-actions bot removed the triage PRs from the community label Jul 1, 2025

lbradstreet reviewed Jul 1, 2025

View reviewed changes

zzbennett added 2 commits July 2, 2025 09:30

cleanup from PR review

0d05da2

Merge branch 'trunk' of github.com:apache/kafka into lbennett-kafka-1…

e9bd51e

…9441

zzbennett requested a review from lbradstreet July 3, 2025 00:33

Merge branch 'trunk' of github.com:apache/kafka into lbennett-kafka-1…

621be71

…9441

zzbennett commented Jul 3, 2025

View reviewed changes

zzbennett added 2 commits July 3, 2025 16:27

remove redundant TopicsDelta interface

27a2df9

slightly enhance metadata delta tests

36055d8

add a javadoc

a2f563e


		Optional<TopicMetadata> topicMetadata(Uuid topicId);

		boolean shareGroupsEnabled();


		CoordinatorMetadataDelta EMPTY = emptyDelta();

		interface TopicsDelta {

KAFKA-19441: encapsulate MetadataImage in GroupCoordinator/ShareCoordinator #20061

Are you sure you want to change the base?

KAFKA-19441: encapsulate MetadataImage in GroupCoordinator/ShareCoordinator #20061

Uh oh!

Conversation

zzbennett commented Jun 28, 2025

Uh oh!

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

zzbennett Jun 30, 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!

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!

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!

zzbennett commented Jun 30, 2025

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!

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!

zzbennett Jun 30, 2025 •

edited

Loading