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/04/07 04:33:48 UTC
[PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Yunyung opened a new pull request, #19397:
URL: https://github.com/apache/kafka/pull/19397
Add [KIP-877](https://cwiki.apache.org/confluence/x/lY3GDQ) support for ConfigProvider.
Cleanup: Collections.singletonMap() -> Map.of()
Jira: https://issues.apache.org/jira/browse/KAFKA-18894
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2034036959
##########
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:
This change is for ConfigTransformer#transform, so we need to update the path where the transform method is used.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on PR #19397:
URL: https://github.com/apache/kafka/pull/19397#issuecomment-3460893585
Yes, I totally agree. We can revisit this if there is a real demand for ConfigProvider. Closing the PR.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2033040494
##########
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:
I don't think we need to wrap `provider` into a `Plugin` at this point, since the metrics are `null` and it doesn't seem to serve any practical purpose.
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java:
##########
@@ -269,18 +270,19 @@ List<String> configProviders() {
Map<String, String> transform(Map<String, String> props) {
// transform worker config according to config.providers
List<String> providerNames = configProviders();
- Map<String, ConfigProvider> providers = new HashMap<>();
+ Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
for (String name : providerNames) {
- ConfigProvider configProvider = plugins.newConfigProvider(
+ Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
this,
- CONFIG_PROVIDERS_CONFIG + "." + name,
- Plugins.ClassLoaderUsage.PLUGINS
+ name,
+ Plugins.ClassLoaderUsage.PLUGINS,
+ null
Review Comment:
ditto: I don't think we need to wrap provider into a Plugin at this point, since the metrics are null and it doesn't seem to serve any practical purpose.
##########
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java:
##########
@@ -587,14 +588,14 @@ private Map<String, Object> configProviderProperties(String configProviderPrefix
* config.providers : A comma-separated list of names for providers.
* config.providers.{name}.class : The Java class name for a provider.
* config.providers.{name}.param.{param-name} : A parameter to be passed to the above Java class on initialization.
- * returns a map of config provider name and its instance.
+ * returns a map of config provider name and its instance wrapped in a {@link org.apache.kafka.common.internals.Plugin}.
*
* @param indirectConfigs The map of potential variable configs
* @param providerConfigProperties The map of config provider configs
* @param classNameFilter Filter for config provider class names
- * @return map of config provider name and its instance.
+ * @return map of config provider name and its instance wrapped in a {@link org.apache.kafka.common.internals.Plugin}.
Review Comment:
We could keep the original document, FYI https://github.com/apache/kafka/pull/19050#discussion_r2031382775
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2045026988
##########
clients/src/main/java/org/apache/kafka/common/config/provider/ConfigProvider.java:
##########
@@ -29,6 +29,10 @@
* <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
Review Comment:
Currently not all config providers can be registered with `PluginMetrics`. Could you please add docs to remind users?
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java:
##########
@@ -269,18 +270,19 @@ List<String> configProviders() {
Map<String, String> transform(Map<String, String> props) {
// transform worker config according to config.providers
List<String> providerNames = configProviders();
- Map<String, ConfigProvider> providers = new HashMap<>();
+ Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
for (String name : providerNames) {
- ConfigProvider configProvider = plugins.newConfigProvider(
+ Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
this,
- CONFIG_PROVIDERS_CONFIG + "." + name,
- Plugins.ClassLoaderUsage.PLUGINS
+ name,
Review Comment:
`CONFIG_PROVIDERS_CONFIG + "." + name` is changed to `name`. Is it expected?
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19397:
URL: https://github.com/apache/kafka/pull/19397#issuecomment-3260306909
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2046927417
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java:
##########
@@ -269,18 +270,19 @@ List<String> configProviders() {
Map<String, String> transform(Map<String, String> props) {
// transform worker config according to config.providers
List<String> providerNames = configProviders();
- Map<String, ConfigProvider> providers = new HashMap<>();
+ Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
for (String name : providerNames) {
- ConfigProvider configProvider = plugins.newConfigProvider(
+ Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
this,
- CONFIG_PROVIDERS_CONFIG + "." + name,
- Plugins.ClassLoaderUsage.PLUGINS
+ name,
Review Comment:
Yes, I improved the newConfigProvider API in order to reuse `name` (providerName) in the method.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2376031213
##########
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:
Not sure I understand this comment. The goal of KAFKA-19149 is to make MirrorSourceConnector and MirrorCheckpointConnector use the new API for their metrics (instead of instantiating their own Metrics), it's not related to ConfigProviders.
##########
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:
I'm not sure I understand your explanation. If we always pass a null Metrics instance, we're never going to call withPluginMetrics(), so ConfigProvider instances can't register any metrics.
##########
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:
We probably want to check `providerPlugin.get()` is not null as well.
##########
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:
Why do we need this class? can't we directly use `MonitorableConfigProvider`?
##########
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:
This assertion is not enough! The test would pass if this code is never executed
##########
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:
Not sure all these changes make sense in this PR. It's a bit noisy and it's not even enough to get rid of the `Collections` import.
##########
clients/src/test/java/org/apache/kafka/common/config/provider/MonitorableConfigProvider.java:
##########
Review Comment:
I don't see any tests in clients that use this class. We should have clients tests using this ConfigProvider and check its metric is registered correctly.
##########
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:
We need to check the metrics from the provider have been created.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2470384579
##########
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 never pass a Metrics instance, how will a ConfigProvider ever be able to register metrics?
> Yes, that's expected, since AbstractConfig doesn't have a Metrics instance.
The point of this work is to figure out a way to do it, or find out it's not possible and decide to not add KIP-877 support to ConfigProvider. Wrapping ConfigProvider into Plugins and never providing metrics is not achieving anything.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2046923569
##########
clients/src/main/java/org/apache/kafka/common/config/provider/ConfigProvider.java:
##########
@@ -29,6 +29,10 @@
* <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
Review Comment:
Sure.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2033053426
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java:
##########
@@ -269,18 +270,19 @@ List<String> configProviders() {
Map<String, String> transform(Map<String, String> props) {
// transform worker config according to config.providers
List<String> providerNames = configProviders();
- Map<String, ConfigProvider> providers = new HashMap<>();
+ Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
for (String name : providerNames) {
- ConfigProvider configProvider = plugins.newConfigProvider(
+ Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
this,
- CONFIG_PROVIDERS_CONFIG + "." + name,
- Plugins.ClassLoaderUsage.PLUGINS
+ name,
+ Plugins.ClassLoaderUsage.PLUGINS,
+ null
Review Comment:
ditto: I don't think we need to wrap `provider` into a `Plugin` at this point, since the metrics are `null` and it doesn't seem to serve any practical purpose.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #19397:
URL: https://github.com/apache/kafka/pull/19397#issuecomment-2783653368
Thanks for this patch, please use `./gradlew spotlessApply` fix build error
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2034037635
##########
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java:
##########
@@ -587,14 +588,14 @@ private Map<String, Object> configProviderProperties(String configProviderPrefix
* config.providers : A comma-separated list of names for providers.
* config.providers.{name}.class : The Java class name for a provider.
* config.providers.{name}.param.{param-name} : A parameter to be passed to the above Java class on initialization.
- * returns a map of config provider name and its instance.
+ * returns a map of config provider name and its instance wrapped in a {@link org.apache.kafka.common.internals.Plugin}.
*
* @param indirectConfigs The map of potential variable configs
* @param providerConfigProperties The map of config provider configs
* @param classNameFilter Filter for config provider class names
- * @return map of config provider name and its instance.
+ * @return map of config provider name and its instance wrapped in a {@link org.apache.kafka.common.internals.Plugin}.
Review Comment:
Sure. I'll update all of them later.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2046928161
##########
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:
Yes, but this change is for ConfigTransformer#transform, we need to update the path where the transform method is used.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on PR #19397:
URL: https://github.com/apache/kafka/pull/19397#issuecomment-3460945698
Thank you for looking into this issue!
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung closed pull request #19397: KAFKA-18894: Add KIP-877 support for ConfigProvider
URL: https://github.com/apache/kafka/pull/19397
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2046928161
##########
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:
This change is for ConfigTransformer#transform, so we need to update the path where the transform method is used.
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java:
##########
@@ -269,18 +270,19 @@ List<String> configProviders() {
Map<String, String> transform(Map<String, String> props) {
// transform worker config according to config.providers
List<String> providerNames = configProviders();
- Map<String, ConfigProvider> providers = new HashMap<>();
+ Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
for (String name : providerNames) {
- ConfigProvider configProvider = plugins.newConfigProvider(
+ Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
this,
- CONFIG_PROVIDERS_CONFIG + "." + name,
- Plugins.ClassLoaderUsage.PLUGINS
+ name,
Review Comment:
This change is for ConfigTransformer#transform, so we need to update the path where the transform method is used.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2034036959
##########
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:
This change is for ConfigTransformer#transform, so we need to update the path where the transform method is used.
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java:
##########
@@ -269,18 +270,19 @@ List<String> configProviders() {
Map<String, String> transform(Map<String, String> props) {
// transform worker config according to config.providers
List<String> providerNames = configProviders();
- Map<String, ConfigProvider> providers = new HashMap<>();
+ Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
for (String name : providerNames) {
- ConfigProvider configProvider = plugins.newConfigProvider(
+ Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
this,
- CONFIG_PROVIDERS_CONFIG + "." + name,
- Plugins.ClassLoaderUsage.PLUGINS
+ name,
+ Plugins.ClassLoaderUsage.PLUGINS,
+ null
Review Comment:
Ditto.
##########
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java:
##########
@@ -587,14 +588,14 @@ private Map<String, Object> configProviderProperties(String configProviderPrefix
* config.providers : A comma-separated list of names for providers.
* config.providers.{name}.class : The Java class name for a provider.
* config.providers.{name}.param.{param-name} : A parameter to be passed to the above Java class on initialization.
- * returns a map of config provider name and its instance.
+ * returns a map of config provider name and its instance wrapped in a {@link org.apache.kafka.common.internals.Plugin}.
*
* @param indirectConfigs The map of potential variable configs
* @param providerConfigProperties The map of config provider configs
* @param classNameFilter Filter for config provider class names
- * @return map of config provider name and its instance.
+ * @return map of config provider name and its instance wrapped in a {@link org.apache.kafka.common.internals.Plugin}.
Review Comment:
Sure. I'll update it later.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on PR #19397:
URL: https://github.com/apache/kafka/pull/19397#issuecomment-3460773805
I agree that monitoring for very short lived instances (clients/brokers) is not very valuable. So adding support for KIP-877 does not really make sense.
As you found, in Connect the ConfigProvider instances can be reused while the Connect runtime is running. In this case monitoring has more value.
However I think it would be confusing to have different behaviors depending on where a ConfigProvider is used. For example if you build a ConfigProvider that has metrics, they would only be emitted when running in Connect and not in other components. I wonder if we should just decide to not add KIP-877 to ConfigProvider. If there is demand for Connect we can reconsider in the future but that does not seem to be the case at the moment. What do you think?
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19397:
URL: https://github.com/apache/kafka/pull/19397#issuecomment-2953136986
@Yunyung could you please sync trunk
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2470357736
##########
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:
I'm not sure your understanding is correct. [KAFKA-19149](https://issues.apache.org/jira/browse/KAFKA-19149) is not about adding metrics to MirrorMaker. MirrorMaker consists of 3 connectors, and 2 of them have metrics. This ticket is to be able to use the new APIs from KIP-877 in these connectors, instead of the hacky way they register metrics today.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #19397:
URL: https://github.com/apache/kafka/pull/19397#discussion_r2470362804
##########
clients/src/test/java/org/apache/kafka/common/config/provider/MonitorableConfigProvider.java:
##########
Review Comment:
ConfigProviders can be used by all Kafka components, Clients (producer, consumer, admin), Brokers, Controllers, Connect and Streams. I'm not sure I understand why you think having tests in WorkerTest is enough.
--
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
Re: [PR] KAFKA-18894: Add KIP-877 support for ConfigProvider [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on PR #19397:
URL: https://github.com/apache/kafka/pull/19397#issuecomment-3458463053
> The point of this work is to figure out a way to do it, or find out it's not possible and decide to not add KIP-877 support to ConfigProvider. Wrapping ConfigProvider into Plugins and never providing metrics is not achieving anything.
> ConfigProviders can be used by all Kafka components, Clients (producer, consumer, admin), Brokers, Controllers, Connect and Streams. I'm not sure I understand why you think having tests in WorkerTest is enough.
@mimaison Thanks for the valuable feedback. After rethinking and checking the code in response to the two comments you mentioned, I found the following:
- For clients (Producer, Consumer, Admin), brokers, controllers, MirrorMaker, and Streams, the ConfigProvider is used only once (Only at startup) in AbstractConfig's constructor and is closed immediately. See: https://github.com/apache/kafka/blob/76a1c839cf1f6e4ac2bc3317a73e65955428a1f6/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java#L562
Therefore, monitoring them are not valuable.
- For Connect, the ConfigProvider is long-lived, enabling dynamic connector configuration updates, config reloads, etc. Monitoring long-lived ConfigProviders is therefore valuable.
Therefore, IMO, KIP-877 support for ConfigProvider is useful only for Connect. WDYT?
--
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