diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala
index e0e991cfbb58f..d2a24fe0bedcf 100644
--- a/core/src/main/scala/kafka/server/ControllerServer.scala
+++ b/core/src/main/scala/kafka/server/ControllerServer.scala
@@ -20,7 +20,7 @@ package kafka.server
 import kafka.network.SocketServer
 import kafka.raft.KafkaRaftManager
 import kafka.server.QuotaFactory.QuotaManagers
-import kafka.server.metadata.{ClientQuotaMetadataManager, DynamicConfigPublisher, KRaftMetadataCachePublisher}
+import kafka.server.metadata.{ClientQuotaMetadataManager, DynamicConfigPublisher}
 
 import scala.collection.immutable
 import kafka.utils.Logging
@@ -35,7 +35,7 @@ import org.apache.kafka.common.utils.internals.LogContext
 import org.apache.kafka.common.{ClusterResource, Endpoint, Uuid}
 import org.apache.kafka.controller.metrics.{ControllerMetadataMetricsPublisher, QuorumControllerMetrics}
 import org.apache.kafka.controller.{Controller, QuorumController, QuorumFeatures}
-import org.apache.kafka.image.publisher.{ControllerRegistrationsPublisher, MetadataPublisher}
+import org.apache.kafka.image.publisher.{ControllerRegistrationsPublisher, KRaftMetadataCachePublisher, MetadataPublisher}
 import org.apache.kafka.metadata.{KafkaConfigSchema, KRaftMetadataCache, ListenerInfo}
 import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer
 import org.apache.kafka.metadata.bootstrap.BootstrapMetadata
diff --git a/core/src/main/scala/kafka/server/metadata/KRaftMetadataCachePublisher.scala b/metadata/src/main/java/org/apache/kafka/image/publisher/KRaftMetadataCachePublisher.java
similarity index 53%
rename from core/src/main/scala/kafka/server/metadata/KRaftMetadataCachePublisher.scala
rename to metadata/src/main/java/org/apache/kafka/image/publisher/KRaftMetadataCachePublisher.java
index 5ba30c05794a6..2b55417f9645c 100644
--- a/core/src/main/scala/kafka/server/metadata/KRaftMetadataCachePublisher.scala
+++ b/metadata/src/main/java/org/apache/kafka/image/publisher/KRaftMetadataCachePublisher.java
@@ -14,25 +14,32 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+package org.apache.kafka.image.publisher;
 
-package kafka.server.metadata
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.image.loader.LoaderManifest;
+import org.apache.kafka.metadata.KRaftMetadataCache;
 
-import org.apache.kafka.image.{MetadataDelta, MetadataImage}
-import org.apache.kafka.image.loader.LoaderManifest
-import org.apache.kafka.image.publisher.MetadataPublisher
-import org.apache.kafka.metadata.KRaftMetadataCache
+public class KRaftMetadataCachePublisher implements MetadataPublisher {
 
-class KRaftMetadataCachePublisher(
-  val metadataCache: KRaftMetadataCache
-) extends MetadataPublisher {
-  override def name(): String = "KRaftMetadataCachePublisher"
+    private final KRaftMetadataCache metadataCache;
 
-  override def onMetadataUpdate(
-    delta: MetadataDelta,
-    newImage: MetadataImage,
-    manifest: LoaderManifest
-  ): Unit = {
-    metadataCache.setImage(newImage)
-  }
-}
+    public KRaftMetadataCachePublisher(KRaftMetadataCache metadataCache) {
+        this.metadataCache = metadataCache;
+    }
+
+    @Override
+    public String name() {
+        return "KRaftMetadataCachePublisher";
+    }
 
+    @Override
+    public void onMetadataUpdate(
+        MetadataDelta delta,
+        MetadataImage newImage,
+        LoaderManifest manifest
+    ) {
+        metadataCache.setImage(newImage);
+    }
+}