Add a test to verify withPluginMetrics is called after configure

Author: Yunyung

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java

@@ -19,10 +19,14 @@
 
 import org.apache.kafka.common.Configurable;
 import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigData;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.provider.ConfigProvider;
 import org.apache.kafka.common.internals.Plugin;
+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.utils.LogCaptureAppender;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.connect.components.Versioned;
@@ -52,6 +56,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLClassLoader;
@@ -398,6 +403,16 @@ public void newConfigProviderShouldConfigureWithPluginClassLoader() {
         assertPluginClassLoaderAlwaysActive(plugin.get());
     }
 
+    @Test
+    public void newConfigProviderShouldCallWithPluginMetricsAfterConfigure() {
+        String providerName = "monitorable";
+        String providerPrefix = WorkerConfig.CONFIG_PROVIDERS_CONFIG + "." + providerName;
+        props.put(providerPrefix + ".class", MonitorableConfigProvider .class.getName());
+        createConfig();
+        Plugin<ConfigProvider> plugin = plugins.newConfigProvider(config, providerName, ClassLoaderUsage.PLUGINS, new Metrics());
+        assertInstanceOf(MonitorableConfigProvider.class, plugin.get());
+    }
+
     @Test
     public void newHeaderConverterShouldConfigureWithPluginClassLoader() {
         props.put(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG, TestPlugin.SAMPLING_HEADER_CONVERTER.className());
@@ -793,4 +808,33 @@ public void configure(Map<String, ?> configs) {
             super.configure(configs);
         }
     }
+
+    public static class MonitorableConfigProvider implements ConfigProvider, Monitorable {
+        private boolean configured = false;
+
+        @Override
+        public void withPluginMetrics(PluginMetrics metrics) {
+            assertTrue(configured);
+        }
+
+        @Override
+        public ConfigData get(String path) {
+            return null;
+        }
+
+        @Override
+        public ConfigData get(String path, Set<String> keys) {
+            return null;
+        }
+
+        @Override
+        public void close() throws IOException {
+        }
+
+        @Override
+        public void configure(Map<String, ?> configs) {
+            configured = true;
+        }
+    }
+
 }