suzhiking commented on code in PR #22586:
URL: https://github.com/apache/kafka/pull/22586#discussion_r3749688926
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1811,75 +1808,49 @@ 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) {
+ if (props.containsKey(config) && !Objects.equals(props.get(config),
streamsValue)) {
+ log.warn(CONTROLLED_CONFIG_OVERRIDE_MESSAGE, clientType, config,
props.get(config), streamsValue);
}
+ props.put(config, streamsValue);
+ }
- if (eosEnabled) {
-
verifyMaxInFlightRequestPerConnection(clientProvidedProps.get(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION));
+ /** Enforce the consumer configs that Streams controls on an
already-assembled consumer config map. */
+ private void enforceControlledConsumerConfigs(final Map<String, Object>
consumerProps) {
+ final Map<String, Object> controlledConfigs =
+ eosEnabled ? CONTROLLED_CONSUMER_CONFIGS_EOS_ENABLED :
CONTROLLED_CONSUMER_CONFIGS;
+ controlledConfigs.forEach((config, streamsValue) ->
+ overwriteControlledConfig(consumerProps, config, streamsValue,
"consumer"));
+ }
+
+ /** Enforce the producer configs that Streams controls on an
already-assembled producer config map. */
+ private void enforceControlledProducerConfigs(final Map<String, Object>
producerProps) {
+ if (!eosEnabled) {
+ return;
}
+ CONTROLLED_PRODUCER_CONFIGS_EOS_ENABLED.forEach((config, streamsValue)
->
+ overwriteControlledConfig(producerProps, config, streamsValue,
"producer"));
+
+ // Streams assigns a unique transactional.id per task later (see
ActiveTaskCreator), so any
+ // user-provided value is ignored. Warn and drop it rather than
forcing a fixed value.
+ if (producerProps.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG))
{
+ log.warn(CONTROLLED_CONFIG_OVERRIDE_MESSAGE, "producer",
ProducerConfig.TRANSACTIONAL_ID_CONFIG,
Review Comment:
Exactly
--
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]
