KAFKA-18891: KIP-877 add support for RemoteLogMetadataManager and RemoteStorageManager

build.gradle

@@ -902,6 +902,7 @@ project(':server') {
     implementation project(':transaction-coordinator')
     implementation project(':raft')
     implementation project(':share-coordinator')
+    implementation project(':storage:storage-api')
     implementation libs.jacksonDatabind
     implementation libs.metrics
     implementation libs.slf4jApi
@@ -913,6 +914,7 @@ project(':server') {
     testImplementation testLog4j2Libs
     testImplementation project(':test-common:test-common-internal-api')
     testImplementation project(':test-common:test-common-runtime')
+    testImplementation project(':storage:storage-api').sourceSets.test.output
 
     testRuntimeOnly runtimeTestLibs
   }

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -220,8 +220,6 @@ class BrokerServer(
         brokerTopicStats,
         logDirFailureChannel)
 
-      remoteLogManagerOpt = createRemoteLogManager()
-
       lifecycleManager = new BrokerLifecycleManager(config,
         time,
         s"broker-${config.nodeId}-",
@@ -280,6 +278,8 @@ class BrokerServer(
             withWildcardHostnamesResolved().
             withEphemeralPortsCorrected(name => socketServer.boundPort(new ListenerName(name)))
 
+      remoteLogManagerOpt = createRemoteLogManager(listenerInfo)
+
       alterPartitionManager = AlterPartitionManager(
         config,
         scheduler = kafkaScheduler,
@@ -471,23 +471,6 @@ class BrokerServer(
         socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
         config.numIoThreads, "RequestHandlerAvgIdlePercent")
 
-      // Start RemoteLogManager before initializing broker metadata publishers.
-      remoteLogManagerOpt.foreach { rlm =>
-        val listenerName = config.remoteLogManagerConfig.remoteLogMetadataManagerListenerName()
-        if (listenerName != null) {
-          val endpoint = listenerInfo.listeners().values().stream
-            .filter(e =>
-              e.listenerName().isPresent &&
-                ListenerName.normalised(e.listenerName().get()).equals(ListenerName.normalised(listenerName))
-            )
-            .findFirst()
-            .orElseThrow(() => new ConfigException(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP,
-              listenerName, "Should be set as a listener name within valid broker listener name list: " + listenerInfo.listeners().values()))
-          rlm.onEndPointCreated(endpoint)
-        }
-        rlm.startup()
-      }
-
       metadataPublishers.add(new MetadataVersionConfigValidator(config, sharedServer.metadataPublishingFaultHandler))
       brokerMetadataPublisher = new BrokerMetadataPublisher(config,
         metadataCache,
@@ -712,16 +695,31 @@ class BrokerServer(
     }
   }
 
-  protected def createRemoteLogManager(): Option[RemoteLogManager] = {
-    if (config.remoteLogManagerConfig.isRemoteStorageSystemEnabled()) {
-      Some(new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
+  protected def createRemoteLogManager(listenerInfo: ListenerInfo): Option[RemoteLogManager] = {
+    if (config.remoteLogManagerConfig.isRemoteStorageSystemEnabled) {
+      val listenerName = config.remoteLogManagerConfig.remoteLogMetadataManagerListenerName()
+      val endpoint = if (listenerName != null) {
+        Some(listenerInfo.listeners().values().stream
+          .filter(e =>
+            e.listenerName().isPresent &&
+              ListenerName.normalised(e.listenerName().get()).equals(ListenerName.normalised(listenerName))
+          )
+          .findFirst()
+          .orElseThrow(() => new ConfigException(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP,
+            listenerName, "Should be set as a listener name within valid broker listener name list: " + listenerInfo.listeners().values())))
+      } else {
+        None
+      }
+
+      val rlm = new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
         (tp: TopicPartition) => logManager.getLog(tp).toJava,
         (tp: TopicPartition, remoteLogStartOffset: java.lang.Long) => {
           logManager.getLog(tp).foreach { log =>
             log.updateLogStartOffsetFromRemoteTier(remoteLogStartOffset)
           }
         },
-        brokerTopicStats, metrics))
+        brokerTopicStats, metrics, endpoint.toJava)
+      Some(rlm)
     } else {
       None
     }

core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala

@@ -2236,7 +2236,8 @@ class UnifiedLogTest {
       _ => Optional.empty[UnifiedLog](),
       (_, _) => {},
       brokerTopicStats,
-      new Metrics()))
+      new Metrics(),
+      Optional.empty))
     remoteLogManager.setDelayedOperationPurgatory(purgatory)
 
     val logConfig = LogTestUtils.createLogConfig(segmentBytes = 200, indexIntervalBytes = 1,
@@ -2333,7 +2334,8 @@ class UnifiedLogTest {
       _ => Optional.empty[UnifiedLog](),
       (_, _) => {},
       brokerTopicStats,
-      new Metrics()))
+      new Metrics(),
+      Optional.empty))
     remoteLogManager.setDelayedOperationPurgatory(purgatory)
 
     val logConfig = LogTestUtils.createLogConfig(segmentBytes = 200, indexIntervalBytes = 1,

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

@@ -3792,8 +3792,8 @@ class ReplicaManagerTest {
       _ => Optional.of(mockLog),
       (TopicPartition, Long) => {},
       brokerTopicStats,
-      metrics)
-    remoteLogManager.startup()
+      metrics,
+      Optional.empty)
     val spyRLM = spy(remoteLogManager)
 
     val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), aliveBrokerIds = Seq(0, 1, 2), enableRemoteStorage = true, shouldMockLog = true, remoteLogManager = Some(spyRLM))
@@ -3903,8 +3903,8 @@ class ReplicaManagerTest {
       _ => Optional.of(dummyLog),
       (TopicPartition, Long) => {},
       brokerTopicStats,
-      metrics)
-    remoteLogManager.startup()
+      metrics,
+      Optional.empty)
     val spyRLM = spy(remoteLogManager)
     val timer = new MockTimer(time)
 

server/src/test/java/org/apache/kafka/server/MonitorablePluginsIntegrationTest.java

@@ -27,43 +27,41 @@
 import org.apache.kafka.common.test.api.ClusterTest;
 import org.apache.kafka.common.test.api.Type;
 import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+import org.apache.kafka.server.log.remote.storage.NoOpRemoteLogMetadataManager;
+import org.apache.kafka.server.log.remote.storage.NoOpRemoteStorageManager;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
 
 import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_SELECTOR_CLASS_CONFIG;
 import static org.apache.kafka.server.config.ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class MonitorablePluginsIntegrationTest {
-
     private static int controllerId(Type type) {
         return type == Type.KRAFT ? 3000 : 0;
     }
 
-    private static Map<String, String> expectedTags(String config, String clazz) {
-        return expectedTags(config, clazz, Map.of());
-    }
-
-    private static Map<String, String> expectedTags(String config, String clazz, Map<String, String> extraTags) {
-        Map<String, String> tags = new LinkedHashMap<>();
-        tags.put("config", config);
-        tags.put("class", clazz);
-        tags.putAll(extraTags);
-        return tags;
-    }
-
     @ClusterTest(
         types = {Type.KRAFT, Type.CO_KRAFT},
         serverProperties = {
             @ClusterConfigProperty(key = StandardAuthorizer.SUPER_USERS_CONFIG, value = "User:ANONYMOUS"),
             @ClusterConfigProperty(key = AUTHORIZER_CLASS_NAME_CONFIG, value = "org.apache.kafka.metadata.authorizer.StandardAuthorizer"),
-            @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableReplicaSelector")
+            @ClusterConfigProperty(key = REPLICA_SELECTOR_CLASS_CONFIG, value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableReplicaSelector"),
+            @ClusterConfigProperty(key = REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, value = "true"),
+            @ClusterConfigProperty(key = REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP,
+                    value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableNoOpRemoteLogMetadataManager"),
+            @ClusterConfigProperty(key = REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP,
+                    value = "org.apache.kafka.server.MonitorablePluginsIntegrationTest$MonitorableNoOpRemoteStorageManager")
         }
     )
     public void testMonitorableServerPlugins(ClusterInstance clusterInstance) {
         assertAuthorizerMetrics(clusterInstance);
         assertReplicaSelectorMetrics(clusterInstance);
+        assertRemoteLogManagerMetrics(clusterInstance);
     }
 
     private void assertAuthorizerMetrics(ClusterInstance clusterInstance) {
@@ -78,6 +76,17 @@ private void assertAuthorizerMetrics(ClusterInstance clusterInstance) {
                 expectedTags(AUTHORIZER_CLASS_NAME_CONFIG, "StandardAuthorizer", Map.of("role", "controller")));
     }
 
+    private void assertRemoteLogManagerMetrics(ClusterInstance clusterInstance) {
+        assertMetrics(
+                clusterInstance.brokers().get(0).metrics(),
+                MonitorableNoOpRemoteLogMetadataManager.METRICS_COUNT,
+                expectedTags(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP, MonitorableNoOpRemoteLogMetadataManager.class.getSimpleName()));
+        assertMetrics(
+                clusterInstance.brokers().get(0).metrics(),
+                MonitorableNoOpRemoteStorageManager.METRICS_COUNT,
+                expectedTags(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, MonitorableNoOpRemoteStorageManager.class.getSimpleName()));
+    }
+
     private void assertReplicaSelectorMetrics(ClusterInstance clusterInstance) {
         assertMetrics(
                 clusterInstance.brokers().get(0).metrics(),
@@ -98,6 +107,17 @@ private void assertMetrics(Metrics metrics, int expected, Map<String, String> ex
         assertEquals(expected, found);
     }
 
+    public static class MonitorableNoOpRemoteLogMetadataManager extends NoOpRemoteLogMetadataManager implements Monitorable {
+
+        private static final int METRICS_COUNT = 1;
+
+        @Override
+        public void withPluginMetrics(PluginMetrics metrics) {
+            MetricName name = metrics.metricName("name", "description", Map.of());
+            metrics.addMetric(name, (Measurable) (config, now) -> 123);
+        }
+    }
+
     public static class MonitorableReplicaSelector extends RackAwareReplicaSelector implements Monitorable {
 
         private static final int METRICS_COUNT = 1;
@@ -108,4 +128,27 @@ public void withPluginMetrics(PluginMetrics metrics) {
             metrics.addMetric(name, (Measurable) (config, now) -> 123);
         }
     }
+
+    public static class MonitorableNoOpRemoteStorageManager extends NoOpRemoteStorageManager implements Monitorable {
+
+        private static final int METRICS_COUNT = 1;
+
+        @Override
+        public void withPluginMetrics(PluginMetrics metrics) {
+            MetricName name = metrics.metricName("name", "description", Map.of());
+            metrics.addMetric(name, (Measurable) (config, now) -> 123);
+        }
+    }
+
+    private static Map<String, String> expectedTags(String config, String clazz) {
+        return expectedTags(config, clazz, Map.of());
+    }
+
+    private static Map<String, String> expectedTags(String config, String clazz, Map<String, String> extraTags) {
+        Map<String, String> tags = new LinkedHashMap<>();
+        tags.put("config", config);
+        tags.put("class", clazz);
+        tags.putAll(extraTags);
+        return tags;
+    }
 }

storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManager.java

@@ -47,6 +47,10 @@
  * "cluster.id", "broker.id" and all other properties prefixed with the config: "remote.log.metadata.manager.impl.prefix"
  * (default value is "rlmm.config.") are passed when {@link #configure(Map)} is invoked on this instance.
  * <p>
+ *
+ * Implement {@link org.apache.kafka.common.metrics.Monitorable} to enable the manager to register metrics.
+ * The following tags are automatically added to all metrics registered: <code>config</code> set to
+ * <code>remote.log.metadata.manager.class.name</code>, and <code>class</code> set to the RemoteLogMetadataManager class name.
  */
 public interface RemoteLogMetadataManager extends Configurable, Closeable {
 

storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageManager.java

@@ -39,6 +39,10 @@
  * <p>
  * All properties prefixed with the config: "remote.log.storage.manager.impl.prefix"
  * (default value is "rsm.config.") are passed when {@link #configure(Map)} is invoked on this instance.
+ *
+ * Implement {@link org.apache.kafka.common.metrics.Monitorable} to enable the manager to register metrics.
+ * The following tags are automatically added to all metrics registered: <code>config</code> set to
+ * <code>remote.log.storage.manager.class.name</code>, and <code>class</code> set to the RemoteStorageManager class name.
  */
 public interface RemoteStorageManager extends Configurable, Closeable {