Copilot commented on code in PR #22586:
URL: https://github.com/apache/kafka/pull/22586#discussion_r4070353178
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1811,77 +1802,66 @@ private Map<String, Object> getCommonConsumerConfigs() {
return consumerProps;
}
- private void checkIfUnexpectedUserSpecifiedClientConfig(final Map<String,
Object> clientProvidedProps,
- final String[]
nonConfigurableConfigs) {
- // Streams does not allow users to configure certain client
configurations (consumer/producer),
- // for example, enable.auto.commit or transactional.id. In cases where
user tries to override
- // such non-configurable client configurations, log a warning and
remove the user defined value
- // from the Map. Thus, the default values for these client
configurations that are suitable for
- // Streams will be used instead.
-
- final String nonConfigurableConfigMessage = "Unexpected user-specified
{} config '{}' found. {} setting ({}) will be ignored and the Streams default
setting ({}) will be used.";
- final String eosMessage = "'" + PROCESSING_GUARANTEE_CONFIG + "' is
set to \"" + getString(PROCESSING_GUARANTEE_CONFIG) + "\". Hence, user";
-
- for (final String config: nonConfigurableConfigs) {
- if (clientProvidedProps.containsKey(config)) {
-
- if (CONSUMER_DEFAULT_OVERRIDES.containsKey(config)) {
- if
(!clientProvidedProps.get(config).equals(CONSUMER_DEFAULT_OVERRIDES.get(config)))
{
- log.error(
- nonConfigurableConfigMessage,
- "consumer",
- config,
- "User",
- clientProvidedProps.get(config),
- CONSUMER_DEFAULT_OVERRIDES.get(config)
- );
- clientProvidedProps.remove(config);
- }
- } else if (eosEnabled) {
- if (CONSUMER_EOS_OVERRIDES.containsKey(config)) {
- if
(!clientProvidedProps.get(config).equals(CONSUMER_EOS_OVERRIDES.get(config))) {
- log.warn(
- nonConfigurableConfigMessage,
- "consumer",
- config,
- eosMessage,
- clientProvidedProps.get(config),
- CONSUMER_EOS_OVERRIDES.get(config)
- );
- clientProvidedProps.remove(config);
- }
- } else if (PRODUCER_EOS_OVERRIDES.containsKey(config)) {
- if
(!clientProvidedProps.get(config).equals(PRODUCER_EOS_OVERRIDES.get(config))) {
- log.warn(
- nonConfigurableConfigMessage,
- "producer",
- config,
- eosMessage,
- clientProvidedProps.get(config),
- PRODUCER_EOS_OVERRIDES.get(config)
- );
- clientProvidedProps.remove(config);
- }
- } else if
(ProducerConfig.TRANSACTIONAL_ID_CONFIG.equals(config)) {
- log.warn(
- nonConfigurableConfigMessage,
- "producer",
- config,
- eosMessage,
- clientProvidedProps.get(config),
- "<appId>-<generatedSuffix>"
- );
- clientProvidedProps.remove(config);
- }
- }
- }
+ private static final String CONTROLLED_CONFIG_OVERRIDE_MESSAGE =
+ "Unexpected user-specified {} config '{}' found. User setting ({})
will be ignored and the Streams default setting ({}) will be used.{}";
Review Comment:
The shared warning says the overridden value will be replaced by the
“Streams default setting,” but this helper is now also used for mandatory
values such as `group.id` and `partition.assignment.strategy`, and the
`transactional.id` path removes the property for later per-task generation
rather than applying this value. This wording can mislead users about what will
actually happen; use wording/action-specific messages for forced versus
generated settings.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1811,77 +1802,66 @@ private Map<String, Object> getCommonConsumerConfigs() {
return consumerProps;
}
- private void checkIfUnexpectedUserSpecifiedClientConfig(final Map<String,
Object> clientProvidedProps,
- final String[]
nonConfigurableConfigs) {
- // Streams does not allow users to configure certain client
configurations (consumer/producer),
- // for example, enable.auto.commit or transactional.id. In cases where
user tries to override
- // such non-configurable client configurations, log a warning and
remove the user defined value
- // from the Map. Thus, the default values for these client
configurations that are suitable for
- // Streams will be used instead.
-
- final String nonConfigurableConfigMessage = "Unexpected user-specified
{} config '{}' found. {} setting ({}) will be ignored and the Streams default
setting ({}) will be used.";
- final String eosMessage = "'" + PROCESSING_GUARANTEE_CONFIG + "' is
set to \"" + getString(PROCESSING_GUARANTEE_CONFIG) + "\". Hence, user";
-
- for (final String config: nonConfigurableConfigs) {
- if (clientProvidedProps.containsKey(config)) {
-
- if (CONSUMER_DEFAULT_OVERRIDES.containsKey(config)) {
- if
(!clientProvidedProps.get(config).equals(CONSUMER_DEFAULT_OVERRIDES.get(config)))
{
- log.error(
- nonConfigurableConfigMessage,
- "consumer",
- config,
- "User",
- clientProvidedProps.get(config),
- CONSUMER_DEFAULT_OVERRIDES.get(config)
- );
- clientProvidedProps.remove(config);
- }
- } else if (eosEnabled) {
- if (CONSUMER_EOS_OVERRIDES.containsKey(config)) {
- if
(!clientProvidedProps.get(config).equals(CONSUMER_EOS_OVERRIDES.get(config))) {
- log.warn(
- nonConfigurableConfigMessage,
- "consumer",
- config,
- eosMessage,
- clientProvidedProps.get(config),
- CONSUMER_EOS_OVERRIDES.get(config)
- );
- clientProvidedProps.remove(config);
- }
- } else if (PRODUCER_EOS_OVERRIDES.containsKey(config)) {
- if
(!clientProvidedProps.get(config).equals(PRODUCER_EOS_OVERRIDES.get(config))) {
- log.warn(
- nonConfigurableConfigMessage,
- "producer",
- config,
- eosMessage,
- clientProvidedProps.get(config),
- PRODUCER_EOS_OVERRIDES.get(config)
- );
- clientProvidedProps.remove(config);
- }
- } else if
(ProducerConfig.TRANSACTIONAL_ID_CONFIG.equals(config)) {
- log.warn(
- nonConfigurableConfigMessage,
- "producer",
- config,
- eosMessage,
- clientProvidedProps.get(config),
- "<appId>-<generatedSuffix>"
- );
- clientProvidedProps.remove(config);
- }
- }
- }
+ private static final String CONTROLLED_CONFIG_OVERRIDE_MESSAGE =
+ "Unexpected user-specified {} config '{}' found. User setting ({})
will be ignored and the Streams default setting ({}) will be used.{}";
+
+ /**
+ * Enforce a config that Streams controls: if the user set a different
value, log a warning that it
+ * is being ignored, then overwrite it with the Streams value. Must be
called after the user-provided
+ * props have been merged into {@code props}, so an override at any prefix
is caught in one place.
+ */
+ private void overwriteControlledConfig(final Map<String, Object> props,
+ final String config,
+ final Object streamsValue,
+ final String clientType) {
+ overwriteControlledConfig(props, config, streamsValue, clientType, "");
+ }
+
+ /**
+ * As {@link #overwriteControlledConfig(Map, String, Object, String)}, but
appends {@code reason} to
+ * the warning to explain why Streams controls the config.
+ */
+ private void overwriteControlledConfig(final Map<String, Object> props,
+ final String config,
+ final Object streamsValue,
+ final String clientType,
+ final String reason) {
+ if (props.containsKey(config) &&
!Objects.equals(String.valueOf(props.get(config)),
String.valueOf(streamsValue))) {
+ log.warn(CONTROLLED_CONFIG_OVERRIDE_MESSAGE, clientType, config,
props.get(config), streamsValue, reason);
}
+ props.put(config, streamsValue);
+ }
+ /** Enforce the consumer configs that Streams controls on an
already-assembled consumer config map. */
+ private void enforceControlledConsumerConfigs(final Map<String, Object>
consumerProps) {
+ CONTROLLED_CONSUMER_CONFIGS.forEach((config, streamsValue) ->
+ overwriteControlledConfig(consumerProps, config, streamsValue,
"consumer"));
Review Comment:
`consumer.group.protocol` is removed from `clientProvidedProps` in
`getCommonConsumerConfigs` before this map is enforced, so a user value
supplied through the generic consumer prefix is silently discarded and never
reaches `overwriteControlledConfig`. That leaves the `group.protocol` entry in
`CONTROLLED_CONSUMER_CONFIGS` without the warning this change is intended to
provide; preserve the prefixed value (while still excluding the top-level
Streams `group.protocol`) until enforcement, or otherwise handle it explicitly.
--
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]
