MINOR: Update `GroupCoordinator` interface to received `MetadataImage/Delta` directly #21008

dajac · 2025-11-27T13:38:57Z

In #20061, we introduced the
CoordinatorMetadataImage and CoordinatorMetadataDelta interfaces to
basically contain/control how metadata is used within the
GroupCoordinatorService, more precisely within the
GroupCoordinatorShard. When we did the change, we directly changed the
GroupCoordinator interface to take implementation of those interfaces,
requiring to wrap the MetadataImage and the MetadataDelta in the
BrokerMetadataPublisher. While looking at it now, I find this not idea
for a couple a reasons:

From a BrokerMetadataPublisher perspective, propagating metadata
should be standardized. Basically, all the internal components should
work the same from his point of view. If a component wants to be more
restrictive within his scope, it is fine but the publisher should not be
aware of this.
From a GroupCoordinator perspective, requiring
CoordinatorMetadataImage and CoordinatorMetadataDelta limits what we
can do. For instance, we could move more functionality such as electing
shards from the BrokerMetadataPublisher to the
GroupCoordinatorService to further simplify the metadata publishing
and improving the encapsulation of the components.
From a ShareCoordinator perspective, the abstraction failed short
as we require CoordinatorMetadataImage and CoordinatorMetadataDelta
in the interface but we still require FeaturesImage as well. The
abstraction fails short here.

This patch is an attempt to change this by moving back to requiring
MetadataImage and MetadataDelta in the GroupCoordinator interface
and to wrap them within the service itself. It does not change the
ShareCoordinator yet. I will do this as a follow-up if people agree
with the general approach.

Reviewers: Sean Quah squah@confluent.io, Lianet Magrans
lmagrans@confluent.io

dajac · 2025-11-27T13:39:51Z

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

+     * This is initialised when the {@link GroupCoordinator#onMetadataUpdate(MetadataDelta, MetadataImage)} is called
     */
-    private CoordinatorMetadataImage metadataImage = null;
+    private volatile CoordinatorMetadataImage metadataImage = null;


This is a bug. The attribute must be volatile because it is accessed from multiple threads.

dajac · 2025-11-27T13:41:17Z

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

+        var wrappedImage = newImage == null ? null : new KRaftCoordinatorMetadataImage(newImage);
+        var wrappedDelta = delta == null ? null : new KRaftCoordinatorMetadataDelta(delta);


This is pretty ugly and it should not be here because the image and the delta can never be null. Unfortunately, we have a couple of tests passing nulls here. I will fix those separately and remove this code in a follow-up. I want the core change to be well scoped.

dajac · 2025-11-27T13:42:02Z

...oordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java

+                metadataImage = new MetadataImageBuilder()
+                    .addTopic(TOPIC_ID, TOPIC_NAME, 1)
+                    .build();


This change is semi-related. It is just easier to build the image rather than mocking it.

squah-confluent

Refactor looks good to me

lianetm

Nice refactoring, makes sense to me. Just a nit.

lianetm · 2025-11-27T21:28:21Z

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

+        var wrappedImage = newImage == null ? null : new KRaftCoordinatorMetadataImage(newImage);
+        var wrappedDelta = delta == null ? null : new KRaftCoordinatorMetadataDelta(delta);
+        metadataImage = wrappedImage;
+        runtime.onNewMetadataImage(wrappedImage, wrappedDelta);


do we want to rename this on the CoordinatorRuntime to onNewMetadataUpdate for consistency?

Our parameter order isn't consistent. We have one onNewMetadataUpdate definition take (delta, image) and the other take (image, delta).

addressed it.

lianetm

Thanks! LGTM.

chia7712

@dajac thanks for this refacor. I have just posted some review comments. Sorry for the delay

chia7712 · 2025-12-01T08:52:46Z

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

        throwIfNotActive();
-        metadataImage = newImage;
-        runtime.onNewMetadataImage(newImage, delta);
+        var wrappedImage = newImage == null ? null : new KRaftCoordinatorMetadataImage(newImage);


Can the image be null? Since a null wrappedImage causes an NPE in CoordinatorRuntime#onMetadataUpdate, if null input is allowed, we should implement a fast-fail

kafka/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

Line 2546 in ed3af72

log.debug("Scheduling applying of a new metadata image with version {}.", newImage.version());

#21008 (comment)

Sorry for neglecting the previous comment. Sounds good

chia7712 · 2025-12-01T08:59:02Z

...tor-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

-        CoordinatorMetadataImage newImage,
-        CoordinatorMetadataDelta delta
+    public void onMetadataUpdate(
+        CoordinatorMetadataDelta delta,


Should the ShareCoordinator also be updated with this change?

From the description:

It does not change the
ShareCoordinator yet. I will do this as a follow-up if people agree
with the general approach

if people agree with the general approach

If the interfaces CoordinatorMetadataImage and CoordinatorMetadataDelta are not referenced by ShareCoordinator or GroupCoordinator, can we remove them by using record class to rewrite them

They are still used by other interfaces so I prefer to keep them as interfaces.

…#21032) This patch is a continuation of #21008. It applies the same naming convention within the `CoordinatorRuntime`. Reviewers: Lianet Magrans <lmagrans@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

…/Delta` directly (#21029) This patch is a continuation of the work started in #21008. It updates the `ShareCoordinator` interface to follow the same pattern. Reviewers: Lianet Magrans <lmagrans@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

…/Delta` directly (apache#21008) In apache#20061, we introduced the `CoordinatorMetadataImage` and `CoordinatorMetadataDelta` interfaces to basically contain/control how metadata is used within the `GroupCoordinatorService`, more precisely within the `GroupCoordinatorShard`. When we did the change, we directly changed the `GroupCoordinator` interface to take implementation of those interfaces, requiring to wrap the `MetadataImage` and the `MetadataDelta` in the `BrokerMetadataPublisher`. While looking at it now, I find this not idea for a couple a reasons: 1) From a `BrokerMetadataPublisher` perspective, propagating metadata should be standardized. Basically, all the internal components should work the same from his point of view. If a component wants to be more restrictive within his scope, it is fine but the publisher should not be aware of this. 2) From a `GroupCoordinator` perspective, requiring `CoordinatorMetadataImage` and `CoordinatorMetadataDelta` limits what we can do. For instance, we could move more functionality such as electing shards from the `BrokerMetadataPublisher` to the `GroupCoordinatorService` to further simplify the metadata publishing and improving the encapsulation of the components. 3) From a `ShareCoordinator` perspective, the abstraction failed short as we require `CoordinatorMetadataImage` and `CoordinatorMetadataDelta` in the interface but we still require `FeaturesImage` as well. The abstraction fails short here. This patch is an attempt to change this by moving back to requiring `MetadataImage` and `MetadataDelta` in the `GroupCoordinator` interface and to wrap them within the service itself. It does not change the `ShareCoordinator` yet. I will do this as a follow-up if people agree with the general approach. Reviewers: Sean Quah <squah@confluent.io>, Lianet Magrans <lmagrans@confluent.io>

dajac added 3 commits November 27, 2025 09:26

onNewMetadataImage -> onMetadataUpdate

0c85625

refactor group coordinator service to use receive a MetadataImage/Delta

d06d36c

cleanup

9c80900

github-actions bot added core Kafka Broker group-coordinator small Small PRs labels Nov 27, 2025

dajac commented Nov 27, 2025

View reviewed changes

dajac changed the title ~~Minor update gc interface~~ MINOR: Update GroupCoordinator interface to received MetadataImage/Delta directly Nov 27, 2025

remove unused import

454a44b

squah-confluent approved these changes Nov 27, 2025

View reviewed changes

lianetm approved these changes Nov 27, 2025

View reviewed changes

address comment

4e830a2

dajac requested a review from lianetm November 28, 2025 08:19

github-actions bot added KIP-932 Queues for Kafka and removed small Small PRs labels Nov 28, 2025

address comment

d090c6e

lianetm approved these changes Nov 28, 2025

View reviewed changes

dajac mentioned this pull request Nov 30, 2025

MINOR: Ensure proper visibility of attribute accesses across threads #21014

Merged

dajac merged commit e9ff2cc into apache:trunk Nov 30, 2025
23 checks passed

dajac deleted the minor-update-gc-interface branch November 30, 2025 08:42

chia7712 reviewed Dec 1, 2025

View reviewed changes

This was referenced Dec 1, 2025

MINOR: Update ShareCoordinator interface to received MetadataImage/Delta directly #21029

Merged

MINOR: Rename CoordinatorShard#onNewMetadataImage to onMetadataUpdate #21032

Merged

		var wrappedImage = newImage == null ? null : new KRaftCoordinatorMetadataImage(newImage);
		var wrappedDelta = delta == null ? null : new KRaftCoordinatorMetadataDelta(delta);

MINOR: Update GroupCoordinator interface to received MetadataImage/Delta directly #21008

MINOR: Update GroupCoordinator interface to received MetadataImage/Delta directly #21008

Uh oh!

Conversation

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

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!

squah-confluent left a comment

Choose a reason for hiding this comment

Uh oh!

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

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

MINOR: Update `GroupCoordinator` interface to received `MetadataImage/Delta` directly #21008

MINOR: Update `GroupCoordinator` interface to received `MetadataImage/Delta` directly #21008

dajac commented Nov 27, 2025 •

edited by github-actions bot

Loading