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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
