You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mimaison (via GitHub)" <gi...@apache.org> on 2026/08/14 16:38:10 UTC
[PR] MINOR: Refactor Connect validation endpoint logic [kafka]
mimaison opened a new pull request, #23163:
URL: https://github.com/apache/kafka/pull/23163
Refactor Connect validation endpoint logic
--
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] MINOR: Refactor Connect validation endpoint logic [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison merged PR #23163:
URL: https://github.com/apache/kafka/pull/23163
--
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] MINOR: Refactor Connect validation endpoint logic [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #23163:
URL: https://github.com/apache/kafka/pull/23163#discussion_r3799905115
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -882,12 +888,53 @@ ConfigInfos validateConnectorConfig(
ConfigInfos clientOverrideInfo = validateClientOverrides(connectorProps, connectorType, connector.getClass(), reportStage, doLog);
ConfigInfos connectorConfigInfo = validateConnectorPluginSpecifiedConfigs(connectorProps, validatedConnectorConfig, enrichedConfigDef, connector, reportStage);
- return mergeConfigInfos(connType,
- connectorConfigInfo,
- clientOverrideInfo,
- converterConfigInfo
- );
+ return redactResolvedValues(
+ mergeConfigInfos(connType, connectorConfigInfo, clientOverrideInfo, converterConfigInfo),
+ originalProps,
+ connectorProps);
+ }
+ }
+
+ static Map<String, String> stripConfigProviderEntries(Map<String, String> connectorProps) {
Review Comment:
Nit: to me this implied stripping all entries that use config provider syntax, instead of all entries that specify custom config providers. Might rename to `removeCustomConfigProviders` or something similar, but up to you
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -882,12 +888,53 @@ ConfigInfos validateConnectorConfig(
ConfigInfos clientOverrideInfo = validateClientOverrides(connectorProps, connectorType, connector.getClass(), reportStage, doLog);
ConfigInfos connectorConfigInfo = validateConnectorPluginSpecifiedConfigs(connectorProps, validatedConnectorConfig, enrichedConfigDef, connector, reportStage);
- return mergeConfigInfos(connType,
- connectorConfigInfo,
- clientOverrideInfo,
- converterConfigInfo
- );
+ return redactResolvedValues(
+ mergeConfigInfos(connType, connectorConfigInfo, clientOverrideInfo, converterConfigInfo),
+ originalProps,
+ connectorProps);
+ }
+ }
+
+ static Map<String, String> stripConfigProviderEntries(Map<String, String> connectorProps) {
+ Map<String, String> sanitized = new HashMap<>(connectorProps);
+ String prefix = AbstractConfig.CONFIG_PROVIDERS_CONFIG + ".";
+ sanitized.keySet().removeIf(k -> k.equals(AbstractConfig.CONFIG_PROVIDERS_CONFIG) || k.startsWith(prefix));
+ return sanitized;
+ }
+
+ static ConfigInfos redactResolvedValues(ConfigInfos infos, Map<String, String> originalProps, Map<String, String> resolvedProps) {
+ Map<String, String> substitutedKeys = new HashMap<>();
+ for (Map.Entry<String, String> entry : originalProps.entrySet()) {
+ String resolvedValue = resolvedProps.get(entry.getKey());
+ if (resolvedValue != null && !resolvedValue.equals(entry.getValue())) {
+ substitutedKeys.put(entry.getKey(), entry.getValue());
+ }
+ }
+ if (substitutedKeys.isEmpty()) {
+ return infos;
+ }
+
+ List<ConfigInfo> redactedConfigs = new ArrayList<>();
+ int errorCount = 0;
+ for (ConfigInfo info : infos.configs()) {
+ ConfigValueInfo cv = info.configValue();
+ if (cv != null && substitutedKeys.containsKey(cv.name())) {
+ String placeholder = substitutedKeys.get(cv.name());
+ List<String> errors = cv.errors().isEmpty()
+ ? cv.errors()
+ : List.of("Invalid value for configuration '" + cv.name()
+ + "': the value resolved from the config provider reference failed validation");
Review Comment:
Can we log the full set of errors at `TRACE` level? Hopefully no one has to resort to this but if they do, better to have that option (along with dynamic logging level adjustments via the REST API) than not
--
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