KAFKA-19441: encapsulate MetadataImage in GroupCoordinator/ShareCoordinator

checkstyle/import-control-coordinator-common.xml

@@ -58,6 +58,7 @@
                 <allow pkg="org.apache.kafka.coordinator.common" />
                 <allow pkg="org.apache.kafka.deferred" />
                 <allow pkg="org.apache.kafka.image" />
+                <allow pkg="org.apache.kafka.metadata" />
                 <allow pkg="org.apache.kafka.server.authorizer" />
                 <allow pkg="org.apache.kafka.server.common" />
                 <allow pkg="org.apache.kafka.server.metrics" />

checkstyle/import-control-jmh-benchmarks.xml

@@ -52,6 +52,8 @@
     <allow pkg="org.apache.kafka.server"/>
     <allow pkg="org.apache.kafka.storage"/>
     <allow pkg="org.apache.kafka.clients"/>
+    <allow class="org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage"/>
+    <allow class="org.apache.kafka.coordinator.common.runtime.KRaftCoordinatorMetadataImage"/>
     <allow class="org.apache.kafka.coordinator.common.runtime.HdrHistogram"/>
     <allow pkg="org.apache.kafka.coordinator.group"/>
     <allow pkg="org.apache.kafka.image"/>

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

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.common.runtime;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Collection;
+import java.util.Set;
+
+/**
+ * Provides metadata deltas to Coordinators (GroupCoordinator, ShareCoordinator, etc) such as changed topics and deleted topics
+ * Implementations should be immutable.
+ */
+public interface CoordinatorMetadataDelta {
+
+    CoordinatorMetadataDelta EMPTY = emptyDelta();
+
+    Collection<Uuid> createdTopicIds();
+
+    Collection<Uuid> changedTopicIds();
+
+    Set<Uuid> deletedTopicIds();
+
+    /**
+     * Returns the previous image of the coordinator metadata.
+     * This image is a snapshot of the metadata before the delta occurred.
+     */
+    CoordinatorMetadataImage image();
+
+    private static CoordinatorMetadataDelta emptyDelta() {
+        return new CoordinatorMetadataDelta() {
+            @Override
+            public Collection<Uuid> createdTopicIds() {
+                return Set.of();
+            }
+
+            @Override
+            public Collection<Uuid> changedTopicIds() {
+                return Set.of();
+            }
+
+            @Override
+            public Set<Uuid> deletedTopicIds() {
+                return Set.of();
+            }
+
+            @Override
+            public CoordinatorMetadataImage image() {
+                return CoordinatorMetadataImage.EMPTY;
+            }
+        };
+    }
+}

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

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.common.runtime;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Provides metadata to Coordinators (GroupCoordinator, ShareCoordinator, etc) such as topics, partitions, and their configurations.
+ * Implementations should be thread-safe and immutable.
+ */
+public interface CoordinatorMetadataImage {
+    static CoordinatorMetadataImage EMPTY = emptyImage();
+
+    Set<Uuid> topicIds();
+
+    Set<String> topicNames();
+
+    Optional<TopicMetadata> topicMetadata(String topicName);
+
+    Optional<TopicMetadata> topicMetadata(Uuid topicId);
+
+    CoordinatorMetadataDelta emptyDelta();
+
+    long version();
+
+    boolean isEmpty();
+
+    /**
+     * Metadata about a particular topic
+     */
+    interface TopicMetadata {
+        String name();
+
+        Uuid id();
+
+        int partitionCount();
+
+        List<String> partitionRacks(int partitionId);
+    }
+
+    private static CoordinatorMetadataImage emptyImage() {
+
+        return new CoordinatorMetadataImage() {
+            @Override
+            public Set<Uuid> topicIds() {
+                return Set.of();
+            }
+
+            @Override
+            public Set<String> topicNames() {
+                return Set.of();
+            }
+
+            @Override
+            public Optional<TopicMetadata> topicMetadata(String topicName) {
+                return Optional.empty();
+            }
+
+            @Override
+            public Optional<TopicMetadata> topicMetadata(Uuid topicId) {
+                return Optional.empty();
+            }
+
+            @Override
+            public CoordinatorMetadataDelta emptyDelta() {
+                return CoordinatorMetadataDelta.EMPTY;
+            }
+
+            @Override
+            public long version() {
+                return 0L;
+            }
+
+            @Override
+            public boolean isEmpty() {
+                return true;
+            }
+        };
+    }
+
+}

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

@@ -39,8 +39,6 @@
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.deferred.DeferredEvent;
 import org.apache.kafka.deferred.DeferredEventQueue;
-import org.apache.kafka.image.MetadataDelta;
-import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.server.util.timer.Timer;
 import org.apache.kafka.server.util.timer.TimerTask;
 import org.apache.kafka.storage.internals.log.LogConfig;
@@ -2009,7 +2007,7 @@ public void onHighWatermarkUpdated(
     /**
      * The latest known metadata image.
      */
-    private volatile MetadataImage metadataImage = MetadataImage.EMPTY;
+    private volatile CoordinatorMetadataImage metadataImage = CoordinatorMetadataImage.EMPTY;
 
     /**
      * Constructor.
@@ -2474,37 +2472,37 @@ public void scheduleUnloadOperation(
      * @param delta     The metadata delta.
      */
     public void onNewMetadataImage(
-        MetadataImage newImage,
-        MetadataDelta delta
+        CoordinatorMetadataImage newImage,
+        CoordinatorMetadataDelta delta
     ) {
         throwIfNotRunning();
-        log.debug("Scheduling applying of a new metadata image with offset {}.", newImage.offset());
+        log.debug("Scheduling applying of a new metadata image with offset {}.", newImage.version());
 
         // Update global image.
         metadataImage = newImage;
 
         // Push an event for each coordinator.
         coordinators.keySet().forEach(tp -> {
-            scheduleInternalOperation("UpdateImage(tp=" + tp + ", offset=" + newImage.offset() + ")", tp, () -> {
+            scheduleInternalOperation("UpdateImage(tp=" + tp + ", offset=" + newImage.version() + ")", tp, () -> {
                 CoordinatorContext context = coordinators.get(tp);
                 if (context != null) {
                     context.lock.lock();
                     try {
                         if (context.state == CoordinatorState.ACTIVE) {
                             // The new image can be applied to the coordinator only if the coordinator
                             // exists and is in the active state.
-                            log.debug("Applying new metadata image with offset {} to {}.", newImage.offset(), tp);
+                            log.debug("Applying new metadata image with offset {} to {}.", newImage.version(), tp);
                             context.coordinator.onNewMetadataImage(newImage, delta);
                         } else {
                             log.debug("Ignored new metadata image with offset {} for {} because the coordinator is not active.",
-                                newImage.offset(), tp);
+                                newImage.version(), tp);
                         }
                     } finally {
                         context.lock.unlock();
                     }
                 } else {
                     log.debug("Ignored new metadata image with offset {} for {} because the coordinator does not exist.",
-                        newImage.offset(), tp);
+                        newImage.version(), tp);
                 }
             });
         });

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

@@ -17,8 +17,6 @@
 package org.apache.kafka.coordinator.common.runtime;
 
 import org.apache.kafka.common.requests.TransactionResult;
-import org.apache.kafka.image.MetadataDelta;
-import org.apache.kafka.image.MetadataImage;
 
 /**
  * CoordinatorShard is basically a replicated state machine managed by the
@@ -32,16 +30,16 @@ public interface CoordinatorShard<U> {
      *
      * @param newImage  The metadata image.
      */
-    default void onLoaded(MetadataImage newImage) {}
+    default void onLoaded(CoordinatorMetadataImage newImage) {}
 
     /**
-     * A new metadata image is available. This is only called after {@link CoordinatorShard#onLoaded(MetadataImage)}
+     * A new metadata image is available. This is only called after {@link CoordinatorShard#onLoaded(CoordinatorMetadataImage)}
      * is called to signal that the coordinator has been fully loaded.
      *
      * @param newImage  The new metadata image.
      * @param delta     The delta image.
      */
-    default void onNewMetadataImage(MetadataImage newImage, MetadataDelta delta) {}
+    default void onNewMetadataImage(CoordinatorMetadataImage newImage, CoordinatorMetadataDelta delta) {}
 
     /**
      * The coordinator has been unloaded. This is used to apply

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

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.common.runtime;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.image.MetadataDelta;
+
+import java.util.Collection;
+import java.util.Set;
+
+/**
+ * An implementation of {@link CoordinatorMetadataDelta} that wraps the KRaft MetadataDelta.
+ */
+public class KRaftCoordinatorMetadataDelta implements CoordinatorMetadataDelta {
+
+    final MetadataDelta metadataDelta;
+
+    public KRaftCoordinatorMetadataDelta(MetadataDelta metadataDelta) {
+        this.metadataDelta = metadataDelta;
+    }
+
+    @Override
+    public Collection<Uuid> createdTopicIds() {
+        if (metadataDelta == null || metadataDelta.topicsDelta() == null) {
+            return Set.of();
+        }
+        return metadataDelta.topicsDelta().createdTopicIds();
+    }
+
+    @Override
+    public Collection<Uuid> changedTopicIds() {
+        if (metadataDelta == null || metadataDelta.topicsDelta() == null) {
+            return Set.of();
+        }
+        return metadataDelta.topicsDelta().changedTopics().keySet();
+    }
+
+    @Override
+    public Set<Uuid> deletedTopicIds() {
+        if (metadataDelta == null || metadataDelta.topicsDelta() == null) {
+            return Set.of();
+        }
+        return metadataDelta.topicsDelta().deletedTopicIds();
+    }
+
+
+    @Override
+    public String toString() {
+        return metadataDelta.toString();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || !o.getClass().equals(this.getClass())) return false;
+        KRaftCoordinatorMetadataDelta other = (KRaftCoordinatorMetadataDelta) o;
+        return metadataDelta.equals(other.metadataDelta);
+    }
+
+    @Override
+    public int hashCode() {
+        return metadataDelta.hashCode();
+    }
+
+    @Override
+    public CoordinatorMetadataImage image() {
+        return new KRaftCoordinatorMetadataImage(metadataDelta.image());
+    }
+}