test tmp

Author: TaiJuWu

core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java

@@ -28,6 +28,8 @@
 import org.apache.kafka.common.errors.ReplicaNotAvailableException;
 import org.apache.kafka.common.metrics.KafkaMetric;
 import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Monitorable;
+import org.apache.kafka.common.metrics.PluginMetrics;
 import org.apache.kafka.common.record.FileRecords;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.RecordBatch;
@@ -3695,9 +3697,9 @@ public void testRLMOpsWhenMetadataIsNotReady() throws InterruptedException, IOEx
             tp -> Optional.of(mockLog),
             (topicPartition, offset) -> currentLogStartOffset.set(offset),
             brokerTopicStats, metrics) {
-            public RemoteStorageManager createRemoteStorageManager() {
-                return remoteStorageManager;
-            }
+//            public RemoteStorageManager createRemoteStorageManager() {
+//                return remoteStorageManager;
+//            }
             public RemoteLogMetadataManager createRemoteLogMetadataManager() {
                 return remoteLogMetadataManager;
             }
@@ -3741,6 +3743,39 @@ long findLogStartOffset(TopicIdPartition topicIdPartition, UnifiedLog log) {
         verifyNoMoreInteractions(remoteStorageManager);
     }
 
+    @Test
+    public void testMonitorableRemoteLogStorageManager() {
+        Properties props = new Properties();
+        props.putAll(brokerConfig);
+        // override common security.protocol by adding "RLMM prefix" and "remote log metadata common client prefix"
+        props.put(DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX + REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX + "security.protocol", "SSL");
+        appendRLMConfig(props);
+        props.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, MonitorableNoOpRemoteStorageManager.class.getName());
+        System.err.println("testMonitorableRemoteLogStorageManager " + props.get(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP));
+        KafkaConfig config = KafkaConfig.fromProps(props);
+        try (RemoteLogManager remoteLogManager = new RemoteLogManager(
+                config.remoteLogManagerConfig(),
+                brokerId,
+                logDir,
+                clusterId,
+                time,
+                tp -> Optional.of(mockLog),
+                (topicPartition, offset) -> { },
+                brokerTopicStats,
+                metrics) {
+//            public RemoteStorageManager createRemoteStorageManager() {
+//                return remoteStorageManager;
+//            }
+            public RemoteLogMetadataManager createRemoteLogMetadataManager() {
+                return remoteLogMetadataManager;
+            }
+        }) {
+            System.err.println("ZZZ " + ((MonitorableNoOpRemoteStorageManager) remoteStorageManager).pluginMetrics);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     private void appendRecordsToFile(File file, int nRecords, int nRecordsPerBatch) throws IOException {
         byte magic = RecordBatch.CURRENT_MAGIC_VALUE;
         Compression compression = Compression.NONE;
@@ -3785,4 +3820,22 @@ private void appendRLMConfig(Properties props) {
         props.put(DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX + remoteLogMetadataProducerTestProp, remoteLogMetadataProducerTestVal);
     }
 
+    public static class MonitorableNoOpRemoteStorageManager extends NoOpRemoteStorageManager implements Monitorable {
+        public boolean pluginMetrics = false;
+
+        public MonitorableNoOpRemoteStorageManager() { }
+
+        @Override
+        public void withPluginMetrics(PluginMetrics metrics) {
+            pluginMetrics = true;
+        }
+    }
+
+    public static class MonitorableNoOpRemoteLogMetadataManager extends NoOpRemoteLogMetadataManager implements Monitorable {
+        public boolean pluginMetrics = false;
+        @Override
+        public void withPluginMetrics(PluginMetrics metrics) {
+            pluginMetrics = true;
+        }
+    }
 }