You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Yunyung (via GitHub)" <gi...@apache.org> on 2025/10/25 03:34:01 UTC
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2461325589
##########
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java:
##########
@@ -620,21 +621,22 @@ private Map<String, ConfigProvider> instantiateConfigProviders(
}
}
// Instantiate Config Providers
- Map<String, ConfigProvider> configProviderInstances = new HashMap<>();
+ Map<String, Plugin<ConfigProvider>> configProviderPluginInstances = new HashMap<>();
for (Map.Entry<String, String> entry : providerMap.entrySet()) {
try {
String prefix = CONFIG_PROVIDERS_CONFIG + "." + entry.getKey() + CONFIG_PROVIDERS_PARAM;
Map<String, ?> configProperties = configProviderProperties(prefix, providerConfigProperties);
ConfigProvider provider = Utils.newInstance(entry.getValue(), ConfigProvider.class);
provider.configure(configProperties);
- configProviderInstances.put(entry.getKey(), provider);
+ Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, CONFIG_PROVIDERS_CONFIG);
Review Comment:
> If we always pass a null Metrics instance, we're never going to call withPluginMetrics(), so ConfigProvider instances can't register any metrics.
Yes, that's expected, since AbstractConfig doesn't have a Metrics instance.
If so, why do we wrap the provider into a Plugin here?
The reason is that the provider will be a member of `ConfigTransformer` and will be used by `ConfigTransformer#transform`.
https://github.com/apache/kafka/pull/19397/files#diff-e1b7dda05dfd0104e51c40c3f3dada978a51e0d734c535a34ca2a0e213edb214R557
```
ConfigTransformer configTransformer = new ConfigTransformer(providerPlugins);
```
So even though we expect it not to register any metrics, we still wrap the `provider` into a `Plugin` here.
This is what I was trying to explain above.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java:
##########
@@ -790,4 +806,13 @@ public void configure(Map<String, ?> configs) {
super.configure(configs);
}
}
+
+ public static class CustomMonitorableConfigProvider extends MonitorableConfigProvider {
+
+ @Override
+ public void withPluginMetrics(PluginMetrics metrics) {
+ assertTrue(configured);
Review Comment:
Great point! Fixed it.
##########
clients/src/test/java/org/apache/kafka/common/config/provider/MonitorableConfigProvider.java:
##########
Review Comment:
I tested it in WorkerTest.java, do you mean that?
##########
clients/src/main/java/org/apache/kafka/common/config/ConfigTransformer.java:
##########
@@ -94,13 +95,13 @@ public ConfigTransformerResult transform(Map<String, String> configs) {
Map<String, Long> ttls = new HashMap<>();
for (Map.Entry<String, Map<String, Set<String>>> entry : keysByProvider.entrySet()) {
String providerName = entry.getKey();
- ConfigProvider provider = configProviders.get(providerName);
+ Plugin<ConfigProvider> providerPlugin = configProviderPlugins.get(providerName);
Map<String, Set<String>> keysByPath = entry.getValue();
- if (provider != null && keysByPath != null) {
+ if (providerPlugin != null && keysByPath != null) {
Review Comment:
Done.
##########
clients/src/main/java/org/apache/kafka/common/config/provider/ConfigProvider.java:
##########
@@ -29,6 +29,11 @@
* <p>Kafka Connect discovers implementations of this interface using the Java {@link java.util.ServiceLoader} mechanism.
* To support this, implementations of this interface should also contain a service provider configuration file in
* {@code META-INF/services/org.apache.kafka.common.config.provider.ConfigProvider}.
+ * <p>Implement {@link org.apache.kafka.common.metrics.Monitorable} to enable the config provider to register metrics.
+ * The following tags are automatically added to all metrics registered: <code>config</code> set to
+ * <code>config.providers</code>, <code>class</code> set to the ConfigProvider class name,
+ * and <code>provider</code> set to the provider name. Note that MirrorMaker will not reflect this behavior until
+ * KAFKA-19149 is implemented.
Review Comment:
My understanding:
MirrorMaker doesn’t have its own Metrics, while MirrorSourceConnector and MirrorCheckpointConnector do. [KAFKA-19149](https://issues.apache.org/jira/browse/KAFKA-19149) aims to update this.
The `ConfigProviderPlugin` needs Metrics from MirrorMaker. Once [KAFKA-19149](https://issues.apache.org/jira/browse/KAFKA-19149) is implemented, these Metrics can be passed properly.
Feel free to correct me if I misunderstood anything.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java:
##########
@@ -383,16 +388,27 @@ public void newConfigProviderShouldConfigureWithPluginClassLoader() {
createConfig();
}
- ConfigProvider plugin = plugins.newConfigProvider(
+ Plugin<ConfigProvider> plugin = plugins.newConfigProvider(
config,
- providerPrefix,
- ClassLoaderUsage.PLUGINS
+ providerName,
+ ClassLoaderUsage.PLUGINS,
+ null
);
- assertInstanceOf(SamplingTestPlugin.class, plugin, "Cannot collect samples");
- Map<String, SamplingTestPlugin> samples = ((SamplingTestPlugin) plugin).flatten();
+ assertInstanceOf(SamplingTestPlugin.class, plugin.get(), "Cannot collect samples");
+ Map<String, SamplingTestPlugin> samples = ((SamplingTestPlugin) plugin.get()).flatten();
assertTrue(samples.containsKey("configure"));
- assertPluginClassLoaderAlwaysActive(plugin);
+ assertPluginClassLoaderAlwaysActive(plugin.get());
+ }
+
+ @Test
+ public void newConfigProviderShouldCallWithPluginMetricsAfterConfigure() {
+ String providerName = "monitorable";
+ String providerPrefix = WorkerConfig.CONFIG_PROVIDERS_CONFIG + "." + providerName;
+ props.put(providerPrefix + ".class", CustomMonitorableConfigProvider.class.getName());
+ createConfig();
+ Plugin<ConfigProvider> plugin = plugins.newConfigProvider(config, providerName, ClassLoaderUsage.PLUGINS, new Metrics());
+ assertInstanceOf(CustomMonitorableConfigProvider.class, plugin.get());
Review Comment:
Done.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java:
##########
@@ -790,4 +806,13 @@ public void configure(Map<String, ?> configs) {
super.configure(configs);
}
}
+
+ public static class CustomMonitorableConfigProvider extends MonitorableConfigProvider {
Review Comment:
Sure. Let's just use `MonitorableConfigProvider` instead.
##########
clients/src/test/java/org/apache/kafka/common/config/ConfigTransformerTest.java:
##########
@@ -45,12 +46,12 @@ public class ConfigTransformerTest {
@BeforeEach
public void setup() {
- configTransformer = new ConfigTransformer(Collections.singletonMap("test", new TestConfigProvider()));
+ configTransformer = new ConfigTransformer(Map.of("test", Plugin.wrapInstance(new TestConfigProvider(), null, "config.providers")));
}
@Test
public void testReplaceVariable() {
- ConfigTransformerResult result = configTransformer.transform(Collections.singletonMap(MY_KEY, "${test:testPath:testKey}"));
+ ConfigTransformerResult result = configTransformer.transform(Map.of(MY_KEY, "${test:testPath:testKey}"));
Review Comment:
Undo the change, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org