MINOR: Update GroupCoordinator interface to received MetadataImage/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>

Author: dajac

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

@@ -2535,12 +2535,12 @@ public void scheduleUnloadOperation(
     /**
      * A new metadata image is available.
      *
-     * @param newImage  The new metadata image.
-     * @param delta     The metadata delta.
+     * @param delta    The metadata delta.
+     * @param newImage The new metadata image.
      */
-    public void onNewMetadataImage(
-        CoordinatorMetadataImage newImage,
-        CoordinatorMetadataDelta delta
+    public void onMetadataUpdate(
+        CoordinatorMetadataDelta delta,
+        CoordinatorMetadataImage newImage
     ) {
         throwIfNotRunning();
         log.debug("Scheduling applying of a new metadata image with version {}.", newImage.version());

coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java

@@ -1968,7 +1968,7 @@ public void testClose() throws Exception {
     }
 
     @Test
-    public void testOnNewMetadataImage() {
+    public void testOnMetadataUpdate() {
         TopicPartition tp0 = new TopicPartition("__consumer_offsets", 0);
         TopicPartition tp1 = new TopicPartition("__consumer_offsets", 1);
 
@@ -2029,7 +2029,7 @@ public void testOnNewMetadataImage() {
         // Publish a new image.
         CoordinatorMetadataDelta delta = new KRaftCoordinatorMetadataDelta(new MetadataDelta(MetadataImage.EMPTY));
         CoordinatorMetadataImage newImage = CoordinatorMetadataImage.EMPTY;
-        runtime.onNewMetadataImage(newImage, delta);
+        runtime.onMetadataUpdate(delta, newImage);
 
         // Coordinator 0 should be notified about it.
         verify(coordinator0).onNewMetadataImage(newImage, delta);

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala

@@ -234,7 +234,7 @@ class BrokerMetadataPublisher(
 
       try {
         // Propagate the new image to the group coordinator.
-        groupCoordinator.onNewMetadataImage(new KRaftCoordinatorMetadataImage(newImage), new KRaftCoordinatorMetadataDelta(delta))
+        groupCoordinator.onMetadataUpdate(delta, newImage)
       } catch {
         case t: Throwable => metadataPublishingFaultHandler.handleFault("Error updating group " +
           s"coordinator with local changes in $deltaName", t)

core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala

@@ -35,7 +35,6 @@ import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.metadata.{FeatureLevelRecord, PartitionRecord, RemoveTopicRecord, TopicRecord}
 import org.apache.kafka.common.test.{KafkaClusterTestKit, TestKitNodes}
 import org.apache.kafka.common.utils.Exit
-import org.apache.kafka.coordinator.common.runtime.{KRaftCoordinatorMetadataDelta, KRaftCoordinatorMetadataImage}
 import org.apache.kafka.coordinator.group.GroupCoordinator
 import org.apache.kafka.coordinator.share.ShareCoordinator
 import org.apache.kafka.image.{AclsImage, ClientQuotasImage, ClusterImageTest, ConfigurationsImage, DelegationTokenImage, FeaturesImage, MetadataDelta, MetadataImage, MetadataImageTest, MetadataProvenance, ProducerIdsImage, ScramImage, TopicsImage}
@@ -292,7 +291,7 @@ class BrokerMetadataPublisherTest {
         .numBytes(42)
         .build())
 
-    verify(groupCoordinator).onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), new KRaftCoordinatorMetadataDelta(delta))
+    verify(groupCoordinator).onMetadataUpdate(delta, image)
   }
 
   @Test

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

@@ -53,9 +53,9 @@
 import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
 import org.apache.kafka.common.requests.TransactionResult;
 import org.apache.kafka.common.utils.BufferSupplier;
-import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataDelta;
-import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
 import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
 
 import java.time.Duration;
@@ -452,12 +452,12 @@ void onResignation(
     /**
      * A new metadata image is available.
      *
-     * @param newImage  The new metadata image.
      * @param delta     The metadata delta.
+     * @param newImage  The new metadata image.
      */
-    void onNewMetadataImage(
-        CoordinatorMetadataImage newImage,
-        CoordinatorMetadataDelta delta
+    void onMetadataUpdate(
+        MetadataDelta delta,
+        MetadataImage newImage
     );
 
     /**

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

@@ -85,18 +85,21 @@
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorEventProcessor;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
-import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataDelta;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorResult;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntimeMetrics;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorShardBuilderSupplier;
+import org.apache.kafka.coordinator.common.runtime.KRaftCoordinatorMetadataDelta;
+import org.apache.kafka.coordinator.common.runtime.KRaftCoordinatorMetadataImage;
 import org.apache.kafka.coordinator.common.runtime.MultiThreadedEventProcessor;
 import org.apache.kafka.coordinator.common.runtime.PartitionWriter;
 import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
 import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
 import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
 import org.apache.kafka.server.authorizer.Authorizer;
 import org.apache.kafka.server.record.BrokerCompressionType;
@@ -344,9 +347,9 @@ public GroupCoordinatorService build() {
 
     /**
      * The metadata image to extract topic id to names map.
-     * This is initialised when the {@link GroupCoordinator#onNewMetadataImage(CoordinatorMetadataImage, CoordinatorMetadataDelta)} is called
+     * This is initialised when the {@link GroupCoordinator#onMetadataUpdate(MetadataDelta, MetadataImage)} is called
      */
-    private CoordinatorMetadataImage metadataImage = null;
+    private volatile CoordinatorMetadataImage metadataImage = null;
 
     /**
      *
@@ -2299,16 +2302,18 @@ public void onResignation(
     }
 
     /**
-     * See {@link GroupCoordinator#onNewMetadataImage(CoordinatorMetadataImage, CoordinatorMetadataDelta)}.
+     * See {@link GroupCoordinator#onMetadataUpdate(MetadataDelta, MetadataImage)}.
      */
     @Override
-    public void onNewMetadataImage(
-        CoordinatorMetadataImage newImage,
-        CoordinatorMetadataDelta delta
+    public void onMetadataUpdate(
+        MetadataDelta delta,
+        MetadataImage newImage
     ) {
         throwIfNotActive();
-        metadataImage = newImage;
-        runtime.onNewMetadataImage(newImage, delta);
+        var wrappedImage = newImage == null ? null : new KRaftCoordinatorMetadataImage(newImage);
+        var wrappedDelta = delta == null ? null : new KRaftCoordinatorMetadataDelta(delta);
+        metadataImage = wrappedImage;
+        runtime.onMetadataUpdate(wrappedDelta, wrappedImage);
     }
 
     /**

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

@@ -88,11 +88,8 @@
 import org.apache.kafka.common.utils.BufferSupplier;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
-import org.apache.kafka.coordinator.common.runtime.KRaftCoordinatorMetadataDelta;
-import org.apache.kafka.coordinator.common.runtime.KRaftCoordinatorMetadataImage;
 import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
 import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
 import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
@@ -3166,7 +3163,7 @@ public void testOnPartitionsDeleted() {
             .addTopic(Uuid.randomUuid(), "foo", 1)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), new KRaftCoordinatorMetadataDelta(new MetadataDelta(image)));
+        service.onMetadataUpdate(new MetadataDelta(image), image);
 
         when(runtime.scheduleWriteAllOperation(
             ArgumentMatchers.eq("on-partition-deleted"),
@@ -3224,7 +3221,7 @@ public void testOnPartitionsDeletedCleanupShareGroupState() {
             .addTopic(Uuid.randomUuid(), "foo", 1)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), new KRaftCoordinatorMetadataDelta(new MetadataDelta(image)));
+        service.onMetadataUpdate(new MetadataDelta(image), image);
 
         // No error in partition deleted callback
         when(runtime.scheduleWriteAllOperation(
@@ -3271,10 +3268,10 @@ public void testOnPartitionsDeletedCleanupShareGroupStateEmptyMetadata() {
             .build();
         service.startup(() -> 3);
 
-        CoordinatorMetadataImage image = new MetadataImageBuilder()
+        MetadataImage image = new MetadataImageBuilder()
             .addTopic(Uuid.randomUuid(), "bar", 1)
-            .buildCoordinatorMetadataImage();
-        service.onNewMetadataImage(image, image.emptyDelta());
+            .build();
+        service.onMetadataUpdate(new MetadataDelta(image), image);
 
         // No error in partition deleted callback
         when(runtime.scheduleWriteAllOperation(
@@ -3321,8 +3318,8 @@ public void testOnPartitionsDeletedCleanupShareGroupStateTopicsNotInMetadata() {
             .build();
         service.startup(() -> 3);
 
-        CoordinatorMetadataImage image = CoordinatorMetadataImage.EMPTY;
-        service.onNewMetadataImage(image, image.emptyDelta());
+        MetadataImage image = MetadataImage.EMPTY;
+        service.onMetadataUpdate(new MetadataDelta(image), image);
 
         // No error in partition deleted callback
         when(runtime.scheduleWriteAllOperation(
@@ -4149,7 +4146,7 @@ public void testDescribeShareGroupAllOffsetsLatestOffsetError() throws Interrupt
             .addTopic(TOPIC_ID, TOPIC_NAME, 3)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(new MetadataDelta(image), image);
 
         int partition = 1;
 
@@ -4232,7 +4229,7 @@ public void testDescribeShareGroupOffsetsMetadataImageNull() throws ExecutionExc
             .build(true);
 
         // Forcing a null Metadata Image
-        service.onNewMetadataImage(null, null);
+        service.onMetadataUpdate(null, null);
 
         int partition = 1;
         DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup requestData = new DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup()
@@ -4276,7 +4273,7 @@ public void testDescribeShareGroupAllOffsets() throws InterruptedException, Exec
             .addTopic(TOPIC_ID, TOPIC_NAME, 3)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(null, image);
 
         int partition = 1;
 
@@ -4347,7 +4344,7 @@ public void testDescribeShareGroupAllOffsetsThrowsError() {
             .addTopic(TOPIC_ID, TOPIC_NAME, 3)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(null, image);
 
         int partition = 1;
 
@@ -4383,7 +4380,7 @@ public void testDescribeShareGroupAllOffsetsNullResult() {
             .addTopic(TOPIC_ID, TOPIC_NAME, 3)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(null, image);
 
         int partition = 1;
 
@@ -4420,7 +4417,7 @@ public void testDescribeShareGroupAllOffsetsReadSummaryPartitionError() throws I
             .addTopic(TOPIC_ID, TOPIC_NAME, 3)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(null, image);
 
         int partition = 1;
 
@@ -4510,7 +4507,7 @@ public void testDescribeShareGroupAllOffsetsMetadataImageNull() throws Execution
             .build(true);
 
         // Forcing a null Metadata Image
-        service.onNewMetadataImage(null, null);
+        service.onMetadataUpdate(null, null);
 
         DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup requestData = new DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup()
             .setGroupId("share-group-id")
@@ -4707,7 +4704,7 @@ public void testDeleteShareGroupOffsetsMetadataImageNull() throws ExecutionExcep
             .build(true);
 
         // Forcing a null Metadata Image
-        service.onNewMetadataImage(null, null);
+        service.onMetadataUpdate(null, null);
 
         DeleteShareGroupOffsetsRequestData requestData = new DeleteShareGroupOffsetsRequestData()
             .setGroupId("share-group-id")
@@ -5584,7 +5581,7 @@ public void testPersisterInitializeSuccess() {
             .addTopic(topicId, "topic-name", 3)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(null, image);
 
         when(mockPersister.initializeState(ArgumentMatchers.any())).thenReturn(CompletableFuture.completedFuture(
             new InitializeShareGroupStateResult.Builder()
@@ -5759,7 +5756,7 @@ public void testPersisterInitializeGroupInitializeFailure() {
             .addTopic(topicId, "topic-name", 3)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(null, image);
 
         when(mockPersister.initializeState(ArgumentMatchers.any())).thenReturn(CompletableFuture.completedFuture(
             new InitializeShareGroupStateResult.Builder()
@@ -5821,7 +5818,7 @@ public void testAlterShareGroupOffsetsMetadataImageNull() throws ExecutionExcept
             .build(true);
 
         // Forcing a null Metadata Image
-        service.onNewMetadataImage(null, null);
+        service.onMetadataUpdate(null, null);
 
         String groupId = "share-group";
         AlterShareGroupOffsetsRequestData request = new AlterShareGroupOffsetsRequestData()
@@ -5977,7 +5974,7 @@ public void testPersisterInitializeForAlterShareGroupOffsetsResponseSuccess() {
             .addTopic(topicId, "topic-name", 1)
             .build();
 
-        service.onNewMetadataImage(new KRaftCoordinatorMetadataImage(image), null);
+        service.onMetadataUpdate(null, image);
 
         when(mockPersister.initializeState(ArgumentMatchers.any())).thenReturn(CompletableFuture.completedFuture(
             new InitializeShareGroupStateResult.Builder()
@@ -6026,7 +6023,7 @@ private static class GroupCoordinatorServiceBuilder {
         private CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime;
         private GroupCoordinatorMetrics metrics = new GroupCoordinatorMetrics();
         private Persister persister = new NoOpStatePersister();
-        private CoordinatorMetadataImage metadataImage = null;
+        private MetadataImage metadataImage = null;
         private PartitionMetadataClient partitionMetadataClient = null;
 
         GroupCoordinatorService build() {
@@ -6035,7 +6032,9 @@ GroupCoordinatorService build() {
 
         GroupCoordinatorService build(boolean serviceStartup) {
             if (metadataImage == null) {
-                metadataImage = mock(CoordinatorMetadataImage.class);
+                metadataImage = new MetadataImageBuilder()
+                    .addTopic(TOPIC_ID, TOPIC_NAME, 1)
+                    .build();
             }
 
             GroupCoordinatorService service = new GroupCoordinatorService(
@@ -6051,14 +6050,8 @@ GroupCoordinatorService build(boolean serviceStartup) {
 
             if (serviceStartup) {
                 service.startup(() -> 1);
-                service.onNewMetadataImage(metadataImage, null);
+                service.onMetadataUpdate(null, metadataImage);
             }
-            when(metadataImage.topicNames()).thenReturn(Set.of(TOPIC_NAME));
-            var topicMetadata = mock(CoordinatorMetadataImage.TopicMetadata.class);
-            when(topicMetadata.name()).thenReturn(TOPIC_NAME);
-            when(topicMetadata.id()).thenReturn(TOPIC_ID);
-            when(metadataImage.topicMetadata(TOPIC_ID)).thenReturn(Optional.of(topicMetadata));
-            when(metadataImage.topicMetadata(TOPIC_NAME)).thenReturn(Optional.of(topicMetadata));
 
             return service;
         }

share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java

@@ -1100,7 +1100,7 @@ public void onTopicsDeleted(Set<Uuid> deletedTopicIds, BufferSupplier bufferSupp
     @Override
     public void onNewMetadataImage(CoordinatorMetadataImage newImage, FeaturesImage newFeaturesImage, CoordinatorMetadataDelta delta) {
         throwIfNotActive();
-        this.runtime.onNewMetadataImage(newImage, delta);
+        this.runtime.onMetadataUpdate(delta, newImage);
         boolean enabled = isShareGroupsEnabled(newFeaturesImage);
         // enabled    shouldRunJob         result (XOR)
         // 0            0               no op on flag, do not call jobs