Clone kafka 18894 #43
Conversation
ConfigProvider Plugin Metrics IntegrationTL;DR: Enhances ConfigProvider with metrics support by wrapping providers in Plugin class for monitoring capabilities. Refacto PR SummaryRefactors ConfigProvider implementation to support metrics collection through the Plugin wrapper class. Change HighlightsClick to expand
Sequence DiagramsequenceDiagram
participant Worker as Worker
participant Plugins as Plugins
participant ConfigProvider as ConfigProvider
participant Plugin as Plugin
participant Metrics as Metrics
Worker->>Plugins: newConfigProvider(config, providerName, ClassLoaderUsage, metrics)
Plugins->>Plugins: Load ConfigProvider class
Plugins->>ConfigProvider: instantiate
Plugins->>ConfigProvider: configure(configs)
Plugins->>Plugin: wrapInstance(provider, metrics, config, tags)
Note over Plugin: Add standard tags:<br>config, class, provider
Plugin->>ConfigProvider: withPluginMetrics(metrics) if Monitorable
Plugins-->>Worker: Return Plugin<ConfigProvider>
Worker->>Worker: Store in providerPluginMap
Testing GuideClick to expand
|
Summary of ChangesHello @DDShantanuBadmanji, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a significant enhancement to how Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
WalkthroughPlugin-wrapped ConfigProvider replaces direct ConfigProvider usage across config resolution paths, adding metrics integration. Constructor signatures and maps update to Map<String, Plugin>. Worker/Plugins now create provider plugins with metrics, and closing logic targets plugins. Tests updated accordingly. One whitespace-only Vagrantfile change; Javadoc added. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor Admin as Admin/Worker
participant Plugins as Plugins
participant Provider as ConfigProvider (impl)
participant Plugin as Plugin<ConfigProvider>
participant Metrics as Metrics
participant WCT as WorkerConfigTransformer
participant CT as ConfigTransformer
Admin->>Plugins: newConfigProvider(config, providerName, classLoaderUsage, metrics)
Plugins->>Provider: instantiate + configure
Plugins->>Plugin: wrapInstance(provider, metrics, tags)
Plugins-->>Admin: Plugin<ConfigProvider>
Admin->>WCT: construct with Map{name -> Plugin<ConfigProvider>}
WCT->>CT: new ConfigTransformer(providerPlugins)
Admin->>CT: transform(configs)
CT->>Plugin: get()
Plugin-->>CT: ConfigProvider
CT->>Provider: get(path, keys)
Provider-->>CT: ConfigData
CT-->>Admin: transformed configs
Admin->>WCT: close()
WCT->>Plugin: close() per provider
sequenceDiagram
autonumber
participant Plugins as Plugins
participant MonProv as MonitorableConfigProvider
participant Pn as Plugin<ConfigProvider>
participant PM as PluginMetrics
Plugins->>MonProv: configure(configs)
Plugins->>Pn: wrapInstance(monProv, metrics, tags)
note over Pn,MonProv: On wrap, withPluginMetrics is invoked
Pn->>MonProv: withPluginMetrics(PluginMetrics)
MonProv->>PM: register metric (config/class/provider tags)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
Pre-merge checks and finishing touches❌ Failed checks (2 warnings, 1 inconclusive)
✨ Finishing touches
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 ast-grep (0.39.5)connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.javaThanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
/refacto-bot |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces metrics for ConfigProvider implementations by wrapping them in a Plugin object, which is a valuable addition for monitoring. The changes are consistently applied across the codebase, including refactoring existing code to use the Plugin<ConfigProvider> wrapper and adding new tests to verify the metrics integration. The code quality is high, and I've found one minor issue regarding metric tag consistency that I've commented on.
| 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); |
There was a problem hiding this comment.
The Plugin wrapper for the ConfigProvider is created without the provider metric tag. The updated Javadoc for ConfigProvider states that the provider tag will be automatically added to all registered metrics. Other parts of the codebase, such as connect.runtime.isolation.Plugins#newConfigProvider, correctly add this tag.
To ensure consistency and adhere to the documented behavior, the provider name, which is available as entry.getKey(), should be included as a tag.
| Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, CONFIG_PROVIDERS_CONFIG); | |
| Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, CONFIG_PROVIDERS_CONFIG, Map.of("provider", entry.getKey())); |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Code Review: ConfigProvider Plugin Integration👍 Well Done
📌 Files Processed
📝 Additional Comments
|
![refacto-test[bot]](https://avatars.githubusercontent.com/u/196302493?s=60&v=4){kind=small}
| plugin.configure(configProviderConfig); | ||
| } | ||
| return plugin; | ||
| return Plugin.wrapInstance(plugin, metrics, WorkerConfig.CONFIG_PROVIDERS_CONFIG, Map.of("provider", providerName)); |
There was a problem hiding this comment.
Resource Leak Risk
The plugin instance is wrapped without closing it if an exception occurs during wrapping. If Plugin.wrapInstance throws an exception, the already configured plugin won't be properly closed, potentially causing resource leaks. This violates proper resource management patterns.
try {
return Plugin.wrapInstance(plugin, metrics, WorkerConfig.CONFIG_PROVIDERS_CONFIG, Map.of("provider", providerName));
} catch (Exception e) {
Utils.closeQuietly(plugin, "config provider");
throw e;
}
Commitable Suggestion
| return Plugin.wrapInstance(plugin, metrics, WorkerConfig.CONFIG_PROVIDERS_CONFIG, Map.of("provider", providerName)); | |
| try { | |
| return Plugin.wrapInstance(plugin, metrics, WorkerConfig.CONFIG_PROVIDERS_CONFIG, Map.of("provider", providerName)); | |
| } catch (Exception e) { | |
| Utils.closeQuietly(plugin, "config provider"); | |
| throw e; | |
| } |
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Resource-Mgmt
- SRE-Error-Handling
| } | ||
| } | ||
| providers.values().forEach(x -> Utils.closeQuietly(x, "config provider")); | ||
| providerPlugins.values().forEach(x -> Utils.closeQuietly(x, "config provider plugin")); |
There was a problem hiding this comment.
Resource Leak Risk
The code attempts to close Plugin objects directly, but Plugin doesn't implement Closeable. This should call x.get() to retrieve the actual ConfigProvider instance before closing. Without this change, ConfigProvider resources won't be properly released.
providerPlugins.values().forEach(x -> Utils.closeQuietly(x.get(), "config provider"));
Commitable Suggestion
| providerPlugins.values().forEach(x -> Utils.closeQuietly(x, "config provider plugin")); | |
| providerPlugins.values().forEach(x -> Utils.closeQuietly(x.get(), "config provider")); |
Standards
- Logic-Verification-Resource-Management
- Algorithm-Correctness-API-Usage
- Business-Rule-Resource-Cleanup
| plugin.configure(configProviderConfig); | ||
| } | ||
| return plugin; | ||
| return Plugin.wrapInstance(plugin, metrics, WorkerConfig.CONFIG_PROVIDERS_CONFIG, Map.of("provider", providerName)); |
There was a problem hiding this comment.
Null Metrics Risk
The metrics parameter can be null (as seen in method signature allowing null) but is passed directly to Plugin.wrapInstance without null checking. This could cause NullPointerException when a ConfigProvider implementing Monitorable tries to use metrics, reducing system reliability.
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Precondition
- SRE-Error-Handling
| public ConfigTransformer(Map<String, Plugin<ConfigProvider>> configProviderPlugins) { | ||
| this.configProviderPlugins = configProviderPlugins; | ||
| } |
There was a problem hiding this comment.
Inconsistent Resource Cleanup
The ConfigTransformer now accepts Plugin instead of ConfigProvider but doesn't update its close method. This inconsistency could lead to improper resource cleanup when ConfigTransformer is closed, potentially causing resource leaks in long-running applications.
Standards
- ISO-IEC-25010-Reliability-Maturity
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- DbC-Resource-Mgmt
- SRE-Resource-Management
| @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"))); |
There was a problem hiding this comment.
Missing Plugin Parameters
The Plugin.wrapInstance() call is missing the extraTags parameter that's used in other similar calls throughout the codebase. This inconsistency could lead to missing provider name tags in metrics, causing monitoring gaps for test providers.
Standards
- Logic-Verification-API-Usage
- Algorithm-Correctness-Parameter-Completeness
| public Plugin<ConfigProvider> newConfigProvider(AbstractConfig config, String providerName, ClassLoaderUsage classLoaderUsage, Metrics metrics) { | ||
| String providerPrefix = WorkerConfig.CONFIG_PROVIDERS_CONFIG + "." + providerName; | ||
| String classPropertyName = providerPrefix + ".class"; |
There was a problem hiding this comment.
Inconsistent Method Signature
Method signature changed from using providerPrefix to providerName, requiring internal reconstruction of the prefix. This creates inconsistency with previous implementation and increases maintenance complexity when caller code needs modification.
Standards
- Clean-Code-Method-Design
- Maintainability-Quality-Consistency
| Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, CONFIG_PROVIDERS_CONFIG); | ||
| configProviderPluginInstances.put(entry.getKey(), providerPlugin); |
There was a problem hiding this comment.
Missing Security Validation
The provider instance is wrapped without validation before usage. Malicious providers could be instantiated and wrapped without proper security checks, potentially allowing untrusted code execution. This creates a security risk when loading external config providers.
Standards
- CWE-20
- OWASP-A03
- NIST-SSDF-PW.1
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (13)
Vagrantfile(1 hunks)clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java(4 hunks)clients/src/main/java/org/apache/kafka/common/config/ConfigTransformer.java(3 hunks)clients/src/main/java/org/apache/kafka/common/config/provider/ConfigProvider.java(1 hunks)clients/src/test/java/org/apache/kafka/common/config/ConfigTransformerTest.java(5 hunks)clients/src/test/java/org/apache/kafka/common/config/provider/MonitorableConfigProvider.java(1 hunks)connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java(2 hunks)connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java(1 hunks)connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfigTransformer.java(3 hunks)connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java(3 hunks)connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConfigTransformerTest.java(5 hunks)connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java(5 hunks)connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java(4 hunks)
🧰 Additional context used
🧬 Code graph analysis (10)
clients/src/test/java/org/apache/kafka/common/config/provider/MonitorableConfigProvider.java (1)
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala (1)
withPluginMetrics(6299-6301)
clients/src/main/java/org/apache/kafka/common/config/ConfigTransformer.java (1)
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfigTransformer.java (2)
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)clients/src/main/java/org/apache/kafka/common/utils/Utils.java (1)
Utils(93-1700)
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java (2)
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java (1)
WorkerConfig(58-463)
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java (3)
clients/src/test/java/org/apache/kafka/common/config/provider/MonitorableConfigProvider.java (1)
MonitorableConfigProvider(29-58)clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java (1)
Metrics(70-696)
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java (2)
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)clients/src/main/java/org/apache/kafka/common/utils/Utils.java (1)
Utils(93-1700)
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConfigTransformerTest.java (1)
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)
clients/src/test/java/org/apache/kafka/common/config/ConfigTransformerTest.java (1)
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java (3)
clients/src/test/java/org/apache/kafka/common/config/provider/MonitorableConfigProvider.java (1)
MonitorableConfigProvider(29-58)clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java (1)
Metrics(70-696)
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java (2)
clients/src/main/java/org/apache/kafka/common/internals/Plugin.java (1)
Plugin(33-92)connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java (1)
Plugins(59-696)
🔇 Additional comments (3)
clients/src/main/java/org/apache/kafka/common/config/ConfigTransformer.java (3)
21-21: LGTM!The import of
Pluginand the field type change fromMap<String, ConfigProvider>toMap<String, Plugin<ConfigProvider>>are correct and align with the broader refactoring to enable metrics integration for config providers.Also applies to: 60-60
65-68: LGTM!The constructor signature and Javadoc updates are correct. The parameter name change from
configProviderstoconfigProviderPluginsimproves clarity and consistency with the new Plugin-based approach.
98-104: LGTM!The changes to retrieve and use the Plugin-wrapped ConfigProvider are correct:
- Line 98: Retrieves
Plugin<ConfigProvider>from the map- Line 100: Null check on the plugin wrapper (correct scope)
- Line 104:
providerPlugin.get().get(path, keys)properly unwraps the plugin to access the ConfigProvider instance before calling itsgetmethodThe logic is sound and maintains the same behavior while enabling metrics support through the Plugin wrapper.
| Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, CONFIG_PROVIDERS_CONFIG); | ||
| configProviderPluginInstances.put(entry.getKey(), providerPlugin); | ||
| } catch (ClassNotFoundException e) { |
There was a problem hiding this comment.
Fix plugin tag key to avoid collapsing metrics
All provider plugins are wrapped with the constant key config.providers, so the resulting metrics tags become identical (config=config.providers, class=<providerSimpleName>). When the same provider class is configured under multiple aliases, their metrics now collide, hiding per-provider signals. Pass the provider alias (or another unique identifier) instead of the constant key so each plugin keeps distinct tags.
Apply this diff:
- Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, CONFIG_PROVIDERS_CONFIG);
+ Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, entry.getKey());🤖 Prompt for AI Agents
In clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
around lines 631 to 633, the plugin wrapper is being created with the constant
CONFIG_PROVIDERS_CONFIG which collapses metrics for different aliases into the
same tag; change the call to Plugin.wrapInstance(...) to pass the provider alias
(e.g., entry.getKey()) or another unique identifier instead of
CONFIG_PROVIDERS_CONFIG so each provider instance keeps distinct metric tags
(ensure any corresponding map keys or subsequent uses expect the alias).
| 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 | ||
| ); | ||
| providers.put(name, configProvider); | ||
| providerPlugins.put(name, configProviderPlugin); | ||
| } |
There was a problem hiding this comment.
Filter out missing config provider plugins
plugins.newConfigProvider(...) returns null when the provider name lacks a <name>.class declaration. By unconditionally inserting that null we hand ConfigTransformer a null plugin (and later try to close it), which leads to an immediate NullPointerException instead of the previous “just skip it” behavior. Please restore the null check before adding to the map.
- providerPlugins.put(name, configProviderPlugin);
+ if (configProviderPlugin != null) {
+ providerPlugins.put(name, configProviderPlugin);
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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 | |
| ); | |
| providers.put(name, configProvider); | |
| providerPlugins.put(name, configProviderPlugin); | |
| } | |
| for (String name : providerNames) { | |
| Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider( | |
| this, | |
| name, | |
| Plugins.ClassLoaderUsage.PLUGINS, | |
| null | |
| ); | |
| if (configProviderPlugin != null) { | |
| providerPlugins.put(name, configProviderPlugin); | |
| } | |
| } |
🤖 Prompt for AI Agents
In
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java
around lines 274 to 282, the loop unconditionally inserts the result of
plugins.newConfigProvider(...) into providerPlugins, but newConfigProvider can
return null when the provider lacks a <name>.class entry; restore a null check
before putting the plugin into providerPlugins (i.e., if the returned
Plugin<ConfigProvider> is null, skip adding it and optionally log a debug/warn
message) so ConfigTransformer never receives a null plugin and later close() is
not invoked on null.
| config, | ||
| providerName, | ||
| ClassLoaderUsage.PLUGINS, | ||
| metrics.metrics() | ||
| ); | ||
| providerMap.put(providerName, configProvider); | ||
| providerPluginMap.put(providerName, configProviderPlugin); | ||
| } |
There was a problem hiding this comment.
Skip null config provider plugins
Plugins.newConfigProvider returns null when the provider name is configured without a corresponding <name>.class. We now insert that null into providerPluginMap, so the first transform attempt dereferences a null plugin and blows up (previous code skipped these). Restore the guard so only successfully-instantiated providers are added.
- providerPluginMap.put(providerName, configProviderPlugin);
+ if (configProviderPlugin != null) {
+ providerPluginMap.put(providerName, configProviderPlugin);
+ }🤖 Prompt for AI Agents
In connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
around lines 215 to 221, the code unconditionally inserts the result of
Plugins.newConfigProvider into providerPluginMap even when that method returns
null; restore the previous null-guard so only non-null configProviderPlugin
instances are put into providerPluginMap (i.e., check if configProviderPlugin !=
null before providerPluginMap.put(...) and otherwise skip/continue, optionally
logging a debug/info message that the named provider had no class configured).
2 participants
Delete this text and replace it with a detailed description of your change. The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are expected
for any behavior change and system tests should be considered for larger
changes.
Summary by CodeRabbit