Nikita-Shupletsov commented on code in PR #22586:
URL: https://github.com/apache/kafka/pull/22586#discussion_r3732221453
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1375,37 +1367,42 @@ public class StreamsConfig extends AbstractConfig {
TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC);
}
- // this is the list of configs for underlying clients
- // that streams prefer different default values
- private static final Map<String, Object> PRODUCER_DEFAULT_OVERRIDES =
Map.of(ProducerConfig.LINGER_MS_CONFIG, "100");
+ // Configs for the underlying clients that Streams prefers different
default values for.
+ // These are only defaults: the user may still override them.
+ private static final Map<String, Object> DEFAULT_CONSUMER_CONFIGS = Map.of(
+ ConsumerConfig.MAX_POLL_RECORDS_CONFIG, DEFAULT_MAX_POLL_RECORDS,
Review Comment:
nit: it's a bit inconsistent, here we have the values as constants, but
bellow(e.g. line 1393) they are inlined.
##########
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:
we don't use overwriteControlledConfig here, because we want to delete the
value completely instead of replacing it with something? or is there any other
reason?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1375,37 +1367,42 @@ public class StreamsConfig extends AbstractConfig {
TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC);
}
- // this is the list of configs for underlying clients
- // that streams prefer different default values
- private static final Map<String, Object> PRODUCER_DEFAULT_OVERRIDES =
Map.of(ProducerConfig.LINGER_MS_CONFIG, "100");
+ // Configs for the underlying clients that Streams prefers different
default values for.
+ // These are only defaults: the user may still override them.
+ private static final Map<String, Object> DEFAULT_CONSUMER_CONFIGS = Map.of(
+ ConsumerConfig.MAX_POLL_RECORDS_CONFIG, DEFAULT_MAX_POLL_RECORDS,
+ ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, DEFAULT_AUTO_OFFSET_RESET
+ );
+
+ private static final Map<String, Object> DEFAULT_PRODUCER_CONFIGS =
Map.of(ProducerConfig.LINGER_MS_CONFIG, DEFAULT_LINGER_MS);
- private static final Map<String, Object> PRODUCER_EOS_OVERRIDES;
+ private static final Map<String, Object>
DEFAULT_PRODUCER_CONFIGS_EOS_ENABLED;
static {
Review Comment:
nit: here we use a static block with Collections.unmodifiableMap, down
bellow we use Map.of.
as Map.of is available for us, I would propose replacing these static blocks
with it
--
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]
