You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2026/09/23 15:38:26 UTC
(kafka) branch trunk updated: KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams (#22586)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 092a8a7de84 KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams (#22586)
092a8a7de84 is described below
commit 092a8a7de845f3aa0cba2e812db0af087e9133c0
Author: Gavin Wang <z2...@uwaterloo.ca>
AuthorDate: Wed Sep 23 11:38:18 2026 -0400
KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams (#22586)
Kafka Streams assigns some of the underlying consumer and producer
configs itself and silently ignored a user-supplied value for them, so
there was no way to tell that the setting had no effect. Two unrelated
mechanisms did this: a checker driven by the `NON_CONFIGURABLE_*` arrays
that inspected the prefix-extracted user properties, and direct `put()`
calls applied after those properties were merged, which warned about
nothing at all.
Replace both with a single helper, `overwriteControlledConfig`, applied
to the fully assembled config map so that an override is caught in one
place whichever prefix supplied it, and split the config sets by intent:
`DEFAULT_*` holds defaults the user may still override, `CONTROLLED_*`
holds values the user cannot. The `_EOS_ONLY` maps carry only what
exactly-once adds and are applied on top of their counterparts.
This closes two gaps. `group.id` and `partition.assignment.strategy` are
documented as controlled by Streams but produced no log line, because
both are forced with a put() after the merge. The per-client prefixes
were only checked against the non-exactly-once set, so
`restore.consumer.isolation.level=read_uncommitted` gave a restore
consumer reading uncommitted records under exactly-once; it now yields
`read_committed`.
Warnings are now all logged at `WARN`, where the non-exactly-once set
was previously logged at `ERROR`, and a config that is only controlled
under exactly-once names the processing guarantee as the reason.
A value that differs from the Streams value only in type, such as
`Boolean.FALSE` where Streams sets `"false"`, no longer warns that it is
being replaced by an identical value.
`max.in.flight.requests.per.connection` is a producer config and is now
validated only there, so a value supplied under a consumer prefix is no
longer rejected.
Reviewers: Nikita Shupletsov <ni...@shupletsov.ca>, Lucas Brutschy
<lb...@confluent.io>, Bill Bejeck <bb...@apache.org>
---
docs/streams/developer-guide/config-streams.md | 43 ++-
.../org/apache/kafka/streams/StreamsConfig.java | 296 ++++++++++-----------
.../apache/kafka/streams/StreamsConfigTest.java | 234 +++++++++++++++-
3 files changed, 402 insertions(+), 171 deletions(-)
diff --git a/docs/streams/developer-guide/config-streams.md b/docs/streams/developer-guide/config-streams.md
index 69f12086ad6..fefc7ef3c2d 100644
--- a/docs/streams/developer-guide/config-streams.md
+++ b/docs/streams/developer-guide/config-streams.md
@@ -1927,7 +1927,7 @@ Producer
### Parameters controlled by Kafka Streams
- Some parameters are not configurable by the user. If you supply a value that is different from the default value, your value is ignored. Below is a list of some of these parameters.
+ Some parameters are not configurable by the user. If you supply a value that is different from the default value, your value is ignored and Kafka Streams logs a warning. Below is a list of some of these parameters.
<table>
<tr>
@@ -1994,8 +1994,36 @@ Consumer
<td>
`StreamsPartitionAssignor`
+</td> </tr>
+ <tr>
+ <td>
+
+group.protocol
+</td>
+ <td>
+
+Consumer
+</td>
+ <td>
+
+`classic`
+</td> </tr>
+ <tr>
+ <td>
+
+bootstrap.resolve.timeout.ms
+</td>
+ <td>
+
+Consumer, Producer, Admin
+</td>
+ <td>
+
+`0`
</td> </tr> </table>
+ The consumer `group.protocol` above is set, for example, as `consumer.group.protocol`. It is distinct from the Kafka Streams `group.protocol` config, which selects the rebalance protocol and can be set to `streams`.
+
If EOS is enabled, other parameters are set with the following values.
<table>
@@ -2037,6 +2065,19 @@ Producer
<td>
`true`
+</td> </tr>
+ <tr>
+ <td>
+
+transactional.id
+</td>
+ <td>
+
+Producer
+</td>
+ <td>
+
+`<application.id>-<processId>-<threadIdx>`
</td> </tr> </table>
### client.id
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index 8c82f8b3128..a17000b5a09 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -67,7 +67,6 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.time.Duration;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -147,6 +146,11 @@ import static org.apache.kafka.common.config.ConfigDef.parseType;
* <li>{@link ConsumerConfig#GROUP_ID_CONFIG "group.id"} (<application.id>) - Streams client will always use the application ID a consumer group ID</li>
* <li>{@link ConsumerConfig#ENABLE_AUTO_COMMIT_CONFIG "enable.auto.commit"} (false) - Streams client will always disable/turn off auto committing</li>
* <li>{@link ConsumerConfig#PARTITION_ASSIGNMENT_STRATEGY_CONFIG "partition.assignment.strategy"} (<code>StreamsPartitionAssignor</code>) - Streams client will always use its own partition assignor</li>
+ * <li>{@link ConsumerConfig#ALLOW_AUTO_CREATE_TOPICS_CONFIG "allow.auto.create.topics"} (false) - Consumers will never trigger automatic topic creation</li>
+ * <li>{@link ConsumerConfig#GROUP_PROTOCOL_CONFIG "group.protocol"} as a consumer config, e.g. {@code consumer.group.protocol} (classic) -
+ * Streams ignores it; to select the rebalance protocol, set the Streams config {@link #GROUP_PROTOCOL_CONFIG "group.protocol"} instead</li>
+ * <li>{@link CommonClientConfigs#BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG "bootstrap.resolve.timeout.ms"} (0) - All clients created by Streams
+ * will always resolve bootstrap servers synchronously</li>
* </ul>
*
* If {@link #PROCESSING_GUARANTEE_CONFIG "processing.guarantee"} is set to {@link #EXACTLY_ONCE_V2 "exactly_once_v2"},
@@ -154,6 +158,8 @@ import static org.apache.kafka.common.config.ConfigDef.parseType;
* <ul>
* <li>{@link ConsumerConfig#ISOLATION_LEVEL_CONFIG "isolation.level"} (read_committed) - Consumers will always read committed data only</li>
* <li>{@link ProducerConfig#ENABLE_IDEMPOTENCE_CONFIG "enable.idempotence"} (true) - Producer will always have idempotency enabled</li>
+ * <li>{@link ProducerConfig#TRANSACTIONAL_ID_CONFIG "transactional.id"} (generated) - Streams sets a unique transactional ID for the
+ * producer of each stream thread</li>
* </ul>
*
* @see KafkaStreams#KafkaStreams(org.apache.kafka.streams.Topology, Properties)
@@ -168,6 +174,8 @@ public class StreamsConfig extends AbstractConfig {
private static final ConfigDef CONFIG;
private final boolean eosEnabled;
+
+ private final String eosControlledConfigReason;
private static final long DEFAULT_COMMIT_INTERVAL_MS = 30000L;
private static final long EOS_DEFAULT_COMMIT_INTERVAL_MS = 100L;
private static final int DEFAULT_TRANSACTION_TIMEOUT = 10000;
@@ -892,20 +900,6 @@ public class StreamsConfig extends AbstractConfig {
@SuppressWarnings("WeakerAccess")
public static final String WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG = "windowstore.changelog.additional.retention.ms";
private static final String WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_DOC = "Added to a windows maintainMs to ensure data is not deleted from the log prematurely. Allows for clock drift. Default is 1 day";
-
- // configs that are locked down for every client Kafka Streams creates (consumers, producer, and admin client)
- private static final String[] NON_CONFIGURABLE_CLIENT_CONFIGS =
- new String[] {CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG};
- private static final String[] NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS =
- new String[] {ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ConsumerConfig.GROUP_PROTOCOL_CONFIG, ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG};
- private static final String[] NON_CONFIGURABLE_CONSUMER_EOS_CONFIGS =
- new String[] {ConsumerConfig.ISOLATION_LEVEL_CONFIG};
- private static final String[] NON_CONFIGURABLE_PRODUCER_EOS_CONFIGS =
- new String[] {
- ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,
- ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,
- ProducerConfig.TRANSACTIONAL_ID_CONFIG
- };
/**
* {@code processing.exception.handler.global.enabled}
* @deprecated Since 4.3. Default will change to {@code true} when removed.
@@ -1378,53 +1372,46 @@ public class StreamsConfig extends AbstractConfig {
TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC);
}
- // overrides that apply to every client Kafka Streams creates; they are merged into the
- // consumer, producer, and admin client override maps below
- private static final Map<String, Object> COMMON_CLIENT_OVERRIDES = Map.of(
- // Kafka Streams does not support asynchronous bootstrap resolution
- CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG, "0"
+ // 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, "1000",
+ ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"
);
- // this is the list of configs for underlying clients
- // that streams prefer different default values
- private static final Map<String, Object> PRODUCER_DEFAULT_OVERRIDES =
- withCommonClientOverrides(Map.of(ProducerConfig.LINGER_MS_CONFIG, "100"));
+ private static final Map<String, Object> DEFAULT_PRODUCER_CONFIGS = Map.of(ProducerConfig.LINGER_MS_CONFIG, "100");
- private static final Map<String, Object> PRODUCER_EOS_OVERRIDES;
- static {
- final Map<String, Object> tempProducerDefaultOverrides = new HashMap<>(PRODUCER_DEFAULT_OVERRIDES);
- tempProducerDefaultOverrides.putAll(Map.of(
- ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE,
- ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true,
- // Reduce the transaction timeout for quicker pending offset expiration on broker side.
- ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, DEFAULT_TRANSACTION_TIMEOUT
- ));
- PRODUCER_EOS_OVERRIDES = Collections.unmodifiableMap(tempProducerDefaultOverrides);
- }
+ // Producer defaults that apply in addition to DEFAULT_PRODUCER_CONFIGS when EOS is enabled.
+ private static final Map<String, Object> DEFAULT_PRODUCER_CONFIGS_EOS_ONLY = Map.of(
+ ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE,
+ // Reduce the transaction timeout for quicker pending offset expiration on broker side.
+ ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, DEFAULT_TRANSACTION_TIMEOUT
+ );
- private static final Map<String, Object> CONSUMER_DEFAULT_OVERRIDES = withCommonClientOverrides(Map.of(
- ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000",
- ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest",
+ // Configs that Streams controls: a user-specified value is ignored (with a warning) and the
+ // Streams value is enforced after the user props have been merged.
+ // Controlled for every client Streams creates: the consumers, the producer, and the admin client.
+ private static final Map<String, Object> CONTROLLED_CLIENT_CONFIGS = Map.of(
+ // Kafka Streams does not support asynchronous bootstrap resolution
+ CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG, "0"
+ );
+
+ private static final Map<String, Object> CONTROLLED_CONSUMER_CONFIGS = Map.of(
ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false",
- ConsumerConfig.GROUP_PROTOCOL_CONFIG, "classic",
- ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "false"
- ));
+ ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "false",
+ ConsumerConfig.GROUP_PROTOCOL_CONFIG, "classic"
+ );
- private static final Map<String, Object> CONSUMER_EOS_OVERRIDES;
- static {
- final Map<String, Object> tempConsumerDefaultOverrides = new HashMap<>(CONSUMER_DEFAULT_OVERRIDES);
- tempConsumerDefaultOverrides.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, READ_COMMITTED.toString());
- CONSUMER_EOS_OVERRIDES = Collections.unmodifiableMap(tempConsumerDefaultOverrides);
- }
+ // Controlled configs that apply in addition to CONTROLLED_CONSUMER_CONFIGS when EOS is enabled.
+ private static final Map<String, Object> CONTROLLED_CONSUMER_CONFIGS_EOS_ONLY =
+ Map.of(ConsumerConfig.ISOLATION_LEVEL_CONFIG, READ_COMMITTED.toString());
- private static final Map<String, Object> ADMIN_CLIENT_OVERRIDES =
- withCommonClientOverrides(Map.of(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG, true));
+ // Controlled producer configs that apply in addition to CONTROLLED_CLIENT_CONFIGS when EOS is enabled.
+ private static final Map<String, Object> CONTROLLED_PRODUCER_CONFIGS_EOS_ONLY =
+ Map.of(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
- private static Map<String, Object> withCommonClientOverrides(final Map<String, Object> overrides) {
- final Map<String, Object> merged = new HashMap<>(COMMON_CLIENT_OVERRIDES);
- merged.putAll(overrides);
- return Collections.unmodifiableMap(merged);
- }
+ private static final Map<String, Object> ADMIN_CLIENT_OVERRIDES =
+ Map.of(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG, true);
public static class InternalConfig {
// This is settable in the main Streams config, but it's a private API for now
@@ -1631,6 +1618,8 @@ public class StreamsConfig extends AbstractConfig {
final boolean doLog) {
super(CONFIG, props, doLog);
eosEnabled = StreamsConfigUtils.eosEnabled(this);
+ eosControlledConfigReason = " Streams controls this config when '" + PROCESSING_GUARANTEE_CONFIG
+ + "' is set to \"" + getString(PROCESSING_GUARANTEE_CONFIG) + "\".";
if (eosEnabled) {
verifyEOSTransactionTimeoutCompatibility();
}
@@ -1808,15 +1797,14 @@ public class StreamsConfig extends AbstractConfig {
}
private Map<String, Object> getCommonConsumerConfigs() {
- final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(CONSUMER_PREFIX, ConsumerConfig.configNames());
-
+ final Map<String, Object> clientProvidedProps = clientProps(ConsumerConfig.configNames(), originals());
+ // The unprefixed group.protocol is the Streams config (classic|streams), not the consumer config of the
+ // same name, so drop it before layering on the consumer-prefixed props. A consumer-prefixed
+ // group.protocol is kept so that it is warned about and overwritten with the other controlled configs.
clientProvidedProps.remove(GROUP_PROTOCOL_CONFIG);
+ clientProvidedProps.putAll(originalsWithPrefix(CONSUMER_PREFIX));
- checkIfUnexpectedUserSpecifiedClientConfig(clientProvidedProps, NON_CONFIGURABLE_CLIENT_CONFIGS, "consumer");
- checkIfUnexpectedUserSpecifiedClientConfig(clientProvidedProps, NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS, "consumer");
- checkIfUnexpectedUserSpecifiedClientConfig(clientProvidedProps, NON_CONFIGURABLE_CONSUMER_EOS_CONFIGS, "consumer");
-
- final Map<String, Object> consumerProps = new HashMap<>(eosEnabled ? CONSUMER_EOS_OVERRIDES : CONSUMER_DEFAULT_OVERRIDES);
+ final Map<String, Object> consumerProps = new HashMap<>(DEFAULT_CONSUMER_CONFIGS);
if (StreamsConfigUtils.eosEnabled(this)) {
consumerProps.put("internal.throw.on.fetch.stable.offset.unsupported", true);
}
@@ -1829,94 +1817,81 @@ public class StreamsConfig extends AbstractConfig {
return consumerProps;
}
- private void checkIfUnexpectedUserSpecifiedClientConfig(final Map<String, Object> clientProvidedProps,
- final String[] nonConfigurableConfigs,
- final String clientType) {
- // Streams does not allow users to configure certain client configurations (consumer/producer/admin),
- // 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 (COMMON_CLIENT_OVERRIDES.containsKey(config)) {
- if (!isSameConfigValue(clientProvidedProps.get(config), COMMON_CLIENT_OVERRIDES.get(config))) {
- log.warn(
- nonConfigurableConfigMessage,
- clientType,
- config,
- "User",
- clientProvidedProps.get(config),
- COMMON_CLIENT_OVERRIDES.get(config)
- );
- clientProvidedProps.remove(config);
- }
- } else if (CONSUMER_DEFAULT_OVERRIDES.containsKey(config)) {
- if (!isSameConfigValue(clientProvidedProps.get(config), 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 (!isSameConfigValue(clientProvidedProps.get(config), 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 (!isSameConfigValue(clientProvidedProps.get(config), 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);
- }
- }
- }
+ // user-provided values may be Strings (e.g. from a Properties file) while the Streams defaults
+ // may be typed, or vice versa; compare their String representations to avoid spurious warnings
+ private static boolean isSameConfigValue(final Object userValue, final Object streamsDefault) {
+ return String.valueOf(userValue).equals(String.valueOf(streamsDefault));
+ }
+
+ private static final String CONTROLLED_CONFIG_OVERRIDE_MESSAGE =
+ "Unexpected user-specified {} config '{}' found. User setting ({}) will be ignored and the value set by Kafka Streams ({}) will be used.{}";
+
+ private static final String TRANSACTIONAL_ID_OVERRIDE_MESSAGE =
+ "Unexpected user-specified producer config '" + ProducerConfig.TRANSACTIONAL_ID_CONFIG + "' found. User setting ({}) will be ignored"
+ + " because Kafka Streams generates a unique " + ProducerConfig.TRANSACTIONAL_ID_CONFIG + " for the producer of each stream thread.{}";
+
+ /**
+ * 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) && !isSameConfigValue(props.get(config), streamsValue)) {
+ log.warn(CONTROLLED_CONFIG_OVERRIDE_MESSAGE, clientType, config, props.get(config), streamsValue, reason);
}
+ props.put(config, streamsValue);
+ }
+ /** Enforce the configs that Streams controls for every client on an already-assembled config map. */
+ private void enforceControlledClientConfigs(final Map<String, Object> clientProps, final String clientType) {
+ CONTROLLED_CLIENT_CONFIGS.forEach((config, streamsValue) ->
+ overwriteControlledConfig(clientProps, config, streamsValue, clientType));
+ }
+
+ /** Enforce the consumer configs that Streams controls on an already-assembled consumer config map. */
+ private void enforceControlledConsumerConfigs(final Map<String, Object> consumerProps) {
+ enforceControlledClientConfigs(consumerProps, "consumer");
+ CONTROLLED_CONSUMER_CONFIGS.forEach((config, streamsValue) ->
+ overwriteControlledConfig(consumerProps, config, streamsValue, "consumer"));
if (eosEnabled) {
- verifyMaxInFlightRequestPerConnection(clientProvidedProps.get(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION));
+ CONTROLLED_CONSUMER_CONFIGS_EOS_ONLY.forEach((config, streamsValue) ->
+ overwriteControlledConfig(consumerProps, config, streamsValue, "consumer", eosControlledConfigReason));
}
}
- // user-provided values may be Strings (e.g. from a Properties file) while the Streams defaults
- // may be typed, or vice versa; compare their String representations to avoid spurious warnings
- private static boolean isSameConfigValue(final Object userValue, final Object streamsDefault) {
- return String.valueOf(userValue).equals(String.valueOf(streamsDefault));
+ /** Enforce the producer configs that Streams controls on an already-assembled producer config map. */
+ private void enforceControlledProducerConfigs(final Map<String, Object> producerProps) {
+ enforceControlledClientConfigs(producerProps, "producer");
+ if (!eosEnabled) {
+ return;
+ }
+ CONTROLLED_PRODUCER_CONFIGS_EOS_ONLY.forEach((config, streamsValue) ->
+ overwriteControlledConfig(producerProps, config, streamsValue, "producer", eosControlledConfigReason));
+
+ // Streams sets a unique transactional.id for each stream thread's producer 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(TRANSACTIONAL_ID_OVERRIDE_MESSAGE,
+ producerProps.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG), eosControlledConfigReason);
+ producerProps.remove(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
+ }
+
+ verifyMaxInFlightRequestPerConnection(producerProps.get(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION));
}
private void verifyMaxInFlightRequestPerConnection(final Object maxInFlightRequests) {
@@ -1961,15 +1936,16 @@ public class StreamsConfig extends AbstractConfig {
// Get main consumer override configs
final Map<String, Object> mainConsumerProps = originalsWithPrefix(MAIN_CONSUMER_PREFIX);
- checkIfUnexpectedUserSpecifiedClientConfig(mainConsumerProps, NON_CONFIGURABLE_CLIENT_CONFIGS, "consumer");
- checkIfUnexpectedUserSpecifiedClientConfig(mainConsumerProps, NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS, "consumer");
consumerProps.putAll(mainConsumerProps);
+ // Enforce the configs Streams controls now that all user-provided consumer props are merged.
+ enforceControlledConsumerConfigs(consumerProps);
+
// this is a hack to work around StreamsConfig constructor inside StreamsPartitionAssignor to avoid casting
consumerProps.put(APPLICATION_ID_CONFIG, groupId);
// add group id, client id with stream client id prefix, and group instance id
- consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ overwriteControlledConfig(consumerProps, ConsumerConfig.GROUP_ID_CONFIG, getString(APPLICATION_ID_CONFIG), "consumer");
consumerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
final String groupInstanceId = (String) consumerProps.get(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG);
// Suffix each thread consumer with thread.id to enforce uniqueness of group.instance.id.
@@ -1978,6 +1954,7 @@ public class StreamsConfig extends AbstractConfig {
}
// add configs required for stream partition assignor
+ overwriteControlledConfig(consumerProps, ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StreamsPartitionAssignor.class.getName(), "consumer");
consumerProps.put(UPGRADE_FROM_CONFIG, getString(UPGRADE_FROM_CONFIG));
consumerProps.put(REPLICATION_FACTOR_CONFIG, getInt(REPLICATION_FACTOR_CONFIG));
consumerProps.put(APPLICATION_SERVER_CONFIG, getString(APPLICATION_SERVER_CONFIG));
@@ -1985,7 +1962,6 @@ public class StreamsConfig extends AbstractConfig {
consumerProps.put(ACCEPTABLE_RECOVERY_LAG_CONFIG, getLong(ACCEPTABLE_RECOVERY_LAG_CONFIG));
consumerProps.put(MAX_WARMUP_REPLICAS_CONFIG, getInt(MAX_WARMUP_REPLICAS_CONFIG));
consumerProps.put(PROBING_REBALANCE_INTERVAL_MS_CONFIG, getLong(PROBING_REBALANCE_INTERVAL_MS_CONFIG));
- consumerProps.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StreamsPartitionAssignor.class.getName());
consumerProps.put(WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG, getLong(WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG));
consumerProps.put(RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG, getInt(RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG));
consumerProps.put(RACK_AWARE_ASSIGNMENT_STRATEGY_CONFIG, getString(RACK_AWARE_ASSIGNMENT_STRATEGY_CONFIG));
@@ -2035,10 +2011,11 @@ public class StreamsConfig extends AbstractConfig {
// Get restore consumer override configs
final Map<String, Object> restoreConsumerProps = originalsWithPrefix(RESTORE_CONSUMER_PREFIX);
- checkIfUnexpectedUserSpecifiedClientConfig(restoreConsumerProps, NON_CONFIGURABLE_CLIENT_CONFIGS, "consumer");
- checkIfUnexpectedUserSpecifiedClientConfig(restoreConsumerProps, NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS, "consumer");
baseConsumerProps.putAll(restoreConsumerProps);
+ // Enforce the configs Streams controls now that all user-provided consumer props are merged.
+ enforceControlledConsumerConfigs(baseConsumerProps);
+
// no need to set group id for a restore consumer
baseConsumerProps.remove(ConsumerConfig.GROUP_ID_CONFIG);
// no need to set instance id for a restore consumer
@@ -2070,10 +2047,11 @@ public class StreamsConfig extends AbstractConfig {
// Get global consumer override configs
final Map<String, Object> globalConsumerProps = originalsWithPrefix(GLOBAL_CONSUMER_PREFIX);
- checkIfUnexpectedUserSpecifiedClientConfig(globalConsumerProps, NON_CONFIGURABLE_CLIENT_CONFIGS, "consumer");
- checkIfUnexpectedUserSpecifiedClientConfig(globalConsumerProps, NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS, "consumer");
baseConsumerProps.putAll(globalConsumerProps);
+ // Enforce the configs Streams controls now that all user-provided consumer props are merged.
+ enforceControlledConsumerConfigs(baseConsumerProps);
+
// no need to set group id for a global consumer
baseConsumerProps.remove(ConsumerConfig.GROUP_ID_CONFIG);
// no need to set instance id for a restore consumer
@@ -2099,14 +2077,17 @@ public class StreamsConfig extends AbstractConfig {
public Map<String, Object> getProducerConfigs(final String clientId) {
final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames());
- checkIfUnexpectedUserSpecifiedClientConfig(clientProvidedProps, NON_CONFIGURABLE_CLIENT_CONFIGS, "producer");
- checkIfUnexpectedUserSpecifiedClientConfig(clientProvidedProps, NON_CONFIGURABLE_PRODUCER_EOS_CONFIGS, "producer");
-
// generate producer configs from original properties and overridden maps
- final Map<String, Object> props = new HashMap<>(eosEnabled ? PRODUCER_EOS_OVERRIDES : PRODUCER_DEFAULT_OVERRIDES);
+ final Map<String, Object> props = new HashMap<>(DEFAULT_PRODUCER_CONFIGS);
+ if (eosEnabled) {
+ props.putAll(DEFAULT_PRODUCER_CONFIGS_EOS_ONLY);
+ }
props.putAll(getClientCustomProps());
props.putAll(clientProvidedProps);
+ // Enforce the configs Streams controls now that all user-provided producer props are merged.
+ enforceControlledProducerConfigs(props);
+
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, originals().get(BOOTSTRAP_SERVERS_CONFIG));
// add client id with stream client id prefix
props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
@@ -2123,12 +2104,13 @@ public class StreamsConfig extends AbstractConfig {
public Map<String, Object> getAdminConfigs(final String clientId) {
final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(ADMIN_CLIENT_PREFIX, AdminClientConfig.configNames());
- checkIfUnexpectedUserSpecifiedClientConfig(clientProvidedProps, NON_CONFIGURABLE_CLIENT_CONFIGS, "admin");
-
final Map<String, Object> props = new HashMap<>(ADMIN_CLIENT_OVERRIDES);
props.putAll(getClientCustomProps());
props.putAll(clientProvidedProps);
+ // Enforce the configs Streams controls now that all user-provided admin props are merged.
+ enforceControlledClientConfigs(props, "admin");
+
// add client id with stream client id prefix
props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index 2b2d4e1f02a..c9cefc97218 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -336,7 +336,7 @@ public class StreamsConfigTest {
final List<String> warnings = appender.getMessages().stream()
.filter(msg -> msg.contains("config 'bootstrap.resolve.timeout.ms' found")
- && msg.contains("User setting (120000) will be ignored and the Streams default setting (0) will be used"))
+ && msg.contains("User setting (120000) will be ignored and the value set by Kafka Streams (0) will be used"))
.collect(Collectors.toList());
assertEquals(5, warnings.size(), "Should log exactly one warning per client, got: " + warnings);
// main, restore, and global consumer
@@ -538,7 +538,7 @@ public class StreamsConfigTest {
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
- appender.setClassLogger(StreamsConfig.class, Level.ERROR);
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
final StreamsConfig streamsConfig = new StreamsConfig(props);
@@ -557,13 +557,13 @@ public class StreamsConfigTest {
assertEquals("false", globalConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
"Global consumer should not allow auto topic creation with consumer.* override");
- // Verify exactly 1 error is logged (consumer.* prefix is validated once in getCommonConsumerConfigs for each type of consumer)
- final List<String> errorMessages = appender.getMessages();
- final long errorCount = errorMessages.stream()
+ // Verify a warning is logged once per consumer (getCommonConsumerConfigs feeds all three consumer types)
+ final List<String> warnMessages = appender.getMessages();
+ final long warnCount = warnMessages.stream()
.filter(msg -> msg.contains("Unexpected user-specified consumer config 'allow.auto.create.topics' found"))
.count();
- assertEquals(3, errorCount,
- "Should log exactly 3 error for consumer.* prefix");
+ assertEquals(3, warnCount,
+ "Should log exactly 3 warnings for consumer.* prefix");
}
}
@@ -575,7 +575,7 @@ public class StreamsConfigTest {
props.put(StreamsConfig.globalConsumerPrefix(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), "true");
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
- appender.setClassLogger(StreamsConfig.class, Level.ERROR);
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
final StreamsConfig streamsConfig = new StreamsConfig(props);
@@ -594,16 +594,78 @@ public class StreamsConfigTest {
assertEquals("false", globalConfigs.get(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG),
"Global consumer should not allow auto topic creation with global.consumer.* override");
- // Verify exactly 3 errors are logged (one for each specific prefix)
- final List<String> errorMessages = appender.getMessages();
- final long errorCount = errorMessages.stream()
+ // Verify exactly 3 warnings are logged (one for each specific prefix)
+ final List<String> warnMessages = appender.getMessages();
+ final long warnCount = warnMessages.stream()
.filter(msg -> msg.contains("Unexpected user-specified consumer config 'allow.auto.create.topics' found"))
.count();
- assertEquals(3, errorCount,
- "Should log exactly 3 errors: one for main.consumer.*, one for restore.consumer.*, one for global.consumer.*");
+ assertEquals(3, warnCount,
+ "Should log exactly 3 warnings: one for main.consumer.*, one for restore.consumer.*, one for global.consumer.*");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"example-application", "dummy", "dummyGroupId"})
+ public void shouldLogWarningAndUseApplicationIdAsGroupIdRegardlessOfRequestedGroupId(final String requestedGroupId) {
+ props.put(consumerPrefix(ConsumerConfig.GROUP_ID_CONFIG), "user-group-id");
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ final Map<String, Object> mainConfigs =
+ streamsConfig.getMainConsumerConfigs(requestedGroupId, clientId, threadIdx);
+
+ assertEquals(groupId, mainConfigs.get(ConsumerConfig.GROUP_ID_CONFIG),
+ "Streams should force group.id to the application id and ignore the user override");
+ assertEquals(1, appender.getMessages().stream()
+ .filter(msg -> msg.contains("Unexpected user-specified consumer config 'group.id' found")
+ && msg.contains(groupId))
+ .count(),
+ "Should log exactly one warning naming the application id as the value Streams will use");
}
}
+ @Test
+ public void shouldLogWarningAndIgnoreUserOverrideOfPartitionAssignmentStrategy() {
+ props.put(consumerPrefix(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG), "com.example.MyAssignor");
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ final Map<String, Object> mainConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
+
+ assertEquals(StreamsPartitionAssignor.class.getName(),
+ mainConfigs.get(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG),
+ "Streams should force its own partition assignor and ignore the user override");
+
+ final long warnCount = appender.getMessages().stream()
+ .filter(msg -> msg.contains("Unexpected user-specified consumer config 'partition.assignment.strategy' found"))
+ .count();
+ assertEquals(1, warnCount, "Should log exactly one warning for the partition.assignment.strategy override");
+ }
+ }
+
+ @Test
+ public void shouldNotLogWarningWhenUserDoesNotOverrideControlledConfigs() {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ // Building each client's configs must not warn about configs the user never set. This guards
+ // against comparing a Streams-seeded default (e.g. auto.offset.reset="earliest") against the
+ // value Streams forces on the restore/global consumers.
+ streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
+ streamsConfig.getRestoreConsumerConfigs(clientId);
+ streamsConfig.getGlobalConsumerConfigs(clientId);
+
+ final long warnCount = appender.getMessages().stream()
+ .filter(msg -> msg.contains("Unexpected user-specified"))
+ .count();
+ assertEquals(0, warnCount, "Should not log a controlled-config warning when the user overrides nothing");
+ }
+ }
@Test
public void shouldSupportNonPrefixedAdminConfigs() {
@@ -751,6 +813,46 @@ public class StreamsConfigTest {
assertEquals("classic", consumerConfigs.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG));
}
+ @Test
+ public void shouldLogWarningWhenUserSetsConsumerPrefixedGroupProtocol() {
+ props.put(StreamsConfig.consumerPrefix(ConsumerConfig.GROUP_PROTOCOL_CONFIG), "consumer");
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
+ streamsConfig.getRestoreConsumerConfigs(clientId);
+ streamsConfig.getGlobalConsumerConfigs(clientId);
+
+ assertEquals(3, appender.getMessages().stream()
+ .filter(msg -> msg.contains("Unexpected user-specified consumer config 'group.protocol' found")
+ && msg.contains("User setting (consumer) will be ignored"))
+ .count(),
+ "Should log one warning per consumer for the consumer-prefixed group.protocol");
+ }
+ }
+
+ @Test
+ public void shouldNotLogConsumerGroupProtocolWarningForStreamsGroupProtocol() {
+ // The unprefixed group.protocol is the Streams config, not the consumer config of the same name.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ assertEquals("classic", streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx).get(ConsumerConfig.GROUP_PROTOCOL_CONFIG));
+ assertEquals("classic", streamsConfig.getRestoreConsumerConfigs(clientId).get(ConsumerConfig.GROUP_PROTOCOL_CONFIG));
+ assertEquals("classic", streamsConfig.getGlobalConsumerConfigs(clientId).get(ConsumerConfig.GROUP_PROTOCOL_CONFIG));
+
+ assertTrue(
+ appender.getMessages().stream().noneMatch(msg -> msg.contains("config 'group.protocol' found")),
+ "Unexpected group.protocol warning(s): " + appender.getMessages()
+ );
+ }
+ }
+
@Test
public void testGetGlobalConsumerConfigsWithGlobalConsumerOverriddenPrefix() {
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "5");
@@ -827,6 +929,62 @@ public class StreamsConfigTest {
assertEquals(READ_COMMITTED.toString(), consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG));
}
+ @Test
+ public void shouldResetToDefaultIfConsumerIsolationLevelIsOverriddenUnderClientPrefixIfEosV2Enabled() {
+ // Controlled configs are enforced on the assembled config map, so an override supplied under a
+ // per-client prefix is caught for every consumer, not just the ones using the generic prefix.
+ props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE_V2);
+ props.put(StreamsConfig.mainConsumerPrefix(ConsumerConfig.ISOLATION_LEVEL_CONFIG), READ_UNCOMMITTED.toString());
+ props.put(StreamsConfig.restoreConsumerPrefix(ConsumerConfig.ISOLATION_LEVEL_CONFIG), READ_UNCOMMITTED.toString());
+ props.put(StreamsConfig.globalConsumerPrefix(ConsumerConfig.ISOLATION_LEVEL_CONFIG), READ_UNCOMMITTED.toString());
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+
+ assertEquals(
+ READ_COMMITTED.toString(),
+ streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx).get(ConsumerConfig.ISOLATION_LEVEL_CONFIG)
+ );
+ assertEquals(
+ READ_COMMITTED.toString(),
+ streamsConfig.getRestoreConsumerConfigs(clientId).get(ConsumerConfig.ISOLATION_LEVEL_CONFIG)
+ );
+ assertEquals(
+ READ_COMMITTED.toString(),
+ streamsConfig.getGlobalConsumerConfigs(clientId).get(ConsumerConfig.ISOLATION_LEVEL_CONFIG)
+ );
+
+ assertEquals(3, appender.getMessages().stream()
+ .filter(msg -> msg.contains("Unexpected user-specified consumer config 'isolation.level' found")
+ && msg.contains("Streams controls this config when 'processing.guarantee' is set to \""
+ + EXACTLY_ONCE_V2 + "\""))
+ .count(),
+ "Should log exactly one warning per consumer, explaining that EOS is the reason");
+ }
+ }
+
+ @Test
+ public void shouldNotLogWarningWhenUserSetsControlledConfigToSameValueWithDifferentType() {
+ // A Boolean false and the "false" that Streams sets are the same setting; warning that the
+ // user's value is being ignored and replaced by an identical one would only confuse them.
+ props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG), false);
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final Map<String, Object> consumerConfigs =
+ new StreamsConfig(props).getMainConsumerConfigs(groupId, clientId, threadIdx);
+
+ assertEquals("false", consumerConfigs.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
+ assertEquals(0, appender.getMessages().stream()
+ .filter(msg -> msg.contains("Unexpected user-specified"))
+ .count(),
+ "Should not warn when the user-specified value only differs in type");
+ }
+ }
+
@Test
public void shouldAllowSettingConsumerIsolationLevelIfEosDisabled() {
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, READ_UNCOMMITTED.toString());
@@ -878,6 +1036,56 @@ public class StreamsConfigTest {
assertNull(producerConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG));
}
+ @Test
+ public void shouldLogWarningAndIgnoreUserOverridesOfControlledProducerConfigsIfEosV2Enabled() {
+ props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE_V2);
+ props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
+ props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "user-TxId");
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ final Map<String, Object> producerConfigs = new StreamsConfig(props).getProducerConfigs(clientId);
+
+ assertTrue((Boolean) producerConfigs.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG),
+ "Streams should force idempotence on under EOS and ignore the user override");
+ assertNull(producerConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG));
+
+ final List<String> messages = appender.getMessages();
+ assertEquals(1, messages.stream()
+ .filter(msg -> msg.contains("Unexpected user-specified producer config 'enable.idempotence' found"))
+ .count(), "Should warn once for the enable.idempotence override");
+ assertEquals(1, messages.stream()
+ .filter(msg -> msg.contains("Unexpected user-specified producer config 'transactional.id' found")
+ && msg.contains("because Kafka Streams generates a unique transactional.id for the producer of each stream thread"))
+ .count(), "Should warn once for the transactional.id override, explaining that Streams generates it");
+ assertEquals(2, messages.stream()
+ .filter(msg -> msg.contains("Streams controls this config when 'processing.guarantee' is set to \""
+ + EXACTLY_ONCE_V2 + "\""))
+ .count(), "Both warnings should explain that the processing guarantee is the reason");
+ }
+ }
+
+ @Test
+ public void shouldNotExplainProcessingGuaranteeForConfigControlledRegardlessOfEosIfEosV2Enabled() {
+ // enable.auto.commit is controlled whether or not EOS is enabled, so the warning must not
+ // attribute it to the processing guarantee even when EOS happens to be on.
+ props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE_V2);
+ props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG), "true");
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ appender.setClassLogger(StreamsConfig.class, Level.WARN);
+
+ new StreamsConfig(props).getMainConsumerConfigs(groupId, clientId, threadIdx);
+
+ assertEquals(1, appender.getMessages().stream()
+ .filter(msg -> msg.contains("Unexpected user-specified consumer config 'enable.auto.commit' found")
+ && !msg.contains("processing.guarantee"))
+ .count(),
+ "The warning must not name the processing guarantee as the reason");
+ }
+ }
+
@Test
public void shouldNotOverrideUserConfigRetriesIfExactlyV2OnceEnabled() {
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE_V2);