Skip to content

KAFKA-13598: enable idempotence producer by default and validate the configs #11691

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 17 commits into from
Feb 5, 2022
Merged

KAFKA-13598: enable idempotence producer by default and validate the configs #11691

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
70b240b
KAFKA-13598: validate producer config for idempotence enabled
showuon Jan 18, 2022
a240180
KAFKA-13598: refactor config.originals
showuon Jan 19, 2022
da25d7b
KAFKA-13598: fix broken tests
showuon Jan 20, 2022
8779dcf
KAFKA-13598: fix broken tests
showuon Jan 20, 2022
a5f23b8
KAFKA-13598: refactor
showuon Jan 20, 2022
ca93d15
KAFKA-13598: fix long hanging test
showuon Jan 21, 2022
14294aa
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-13598
showuon Jan 21, 2022
4d5d527
KAFKA-13598: test
showuon Jan 21, 2022
3c2f1de
KAFKA-13598: address reviewer's comments to refactor the code
showuon Jan 26, 2022
a3c7bcf
KAFKA-13598: fix flaky tests and typo
showuon Jan 28, 2022
03cd033
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-13598
showuon Jan 28, 2022
eadc897
KAFKA-13598: fix compile error
showuon Jan 28, 2022
1a96c6b
KAFKA-13598: remove the change in AbstractConfig and refactor
showuon Jan 29, 2022
d1e3bcb
KAFKA-13598: add a notable change for 3.2.0
showuon Jan 29, 2022
d312fe0
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-13598
showuon Jan 30, 2022
306501a
KAFKA-13598: add tests with transactionID + default idempotence config
showuon Feb 4, 2022
9ba42d1
KAFKA-13598: refactor upgrade.doc
showuon Feb 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,8 +194,8 @@ public class CommonClientConfigs {
public static Map<String, Object> postProcessReconnectBackoffConfigs(AbstractConfig config,
Map<String, Object> parsedValues) {
HashMap<String, Object> rval = new HashMap<>();
if ((!config.originals().containsKey(RECONNECT_BACKOFF_MAX_MS_CONFIG)) &&
config.originals().containsKey(RECONNECT_BACKOFF_MS_CONFIG)) {
if ((!config.hasKeyInOriginals(RECONNECT_BACKOFF_MAX_MS_CONFIG)) &&
config.hasKeyInOriginals(RECONNECT_BACKOFF_MS_CONFIG)) {

@showuon showuon Jan 19, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. refactor the originals().containsKey usage. In originals(), we'll make a copy of the configs, and in most cases, we only want to check containsKey. Refactor it by directly check the key in AbstractConfigs to avoid unnecessary map copy.

log.debug("Disabling exponential reconnect backoff because {} is set, but {} is not.",
RECONNECT_BACKOFF_MS_CONFIG, RECONNECT_BACKOFF_MAX_MS_CONFIG);
rval.put(RECONNECT_BACKOFF_MAX_MS_CONFIG, parsedValues.get(RECONNECT_BACKOFF_MS_CONFIG));
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -608,7 +608,7 @@ private int configureDefaultApiTimeoutMs(AdminClientConfig config) {
int defaultApiTimeoutMs = config.getInt(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);

if (defaultApiTimeoutMs < requestTimeoutMs) {
if (config.originals().containsKey(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG)) {
if (config.hasKeyInOriginals(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG)) {
throw new ConfigException("The specified value of " + AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG +
" must be no smaller than the value of " + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG + ".");
} else {
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -613,7 +613,7 @@ boolean maybeOverrideEnableAutoCommit() {
Optional<String> groupId = Optional.ofNullable(getString(CommonClientConfigs.GROUP_ID_CONFIG));
boolean enableAutoCommit = getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
if (!groupId.isPresent()) { // overwrite in case of default group id where the config is not explicitly provided
if (!originals().containsKey(ENABLE_AUTO_COMMIT_CONFIG)) {
if (!hasKeyInOriginals(ENABLE_AUTO_COMMIT_CONFIG)) {
enableAutoCommit = false;
} else if (enableAutoCommit) {
throw new InvalidConfigurationException(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG + " cannot be set to true when default group id (null) is used.");
Expand Down
36 changes: 4 additions & 32 deletions clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -442,7 +442,7 @@ public KafkaProducer(Properties properties, Serializer<K> keySerializer, Seriali

// visible for testing
Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {
int maxInflightRequests = configureInflightRequests(producerConfig);
int maxInflightRequests = producerConfig.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
int requestTimeoutMs = producerConfig.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG);
ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(producerConfig, time, logContext);
ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);
Expand All @@ -465,7 +465,8 @@ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadat
apiVersions,
throttleTimeSensor,
logContext);
short acks = configureAcks(producerConfig, log);

short acks = Short.parseShort(producerConfig.getString(ProducerConfig.ACKS_CONFIG));
return new Sender(logContext,
client,
metadata,
Expand Down Expand Up @@ -493,7 +494,7 @@ private static int configureDeliveryTimeout(ProducerConfig config, Logger log) {
int lingerAndRequestTimeoutMs = (int) Math.min((long) lingerMs + requestTimeoutMs, Integer.MAX_VALUE);

if (deliveryTimeoutMs < lingerAndRequestTimeoutMs) {
if (config.originals().containsKey(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG)) {
if (config.hasKeyInOriginals(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG)) {
// throw an exception if the user explicitly set an inconsistent value
throw new ConfigException(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG
+ " should be equal to or larger than " + ProducerConfig.LINGER_MS_CONFIG
Expand All @@ -511,15 +512,8 @@ private static int configureDeliveryTimeout(ProducerConfig config, Logger log) {

private TransactionManager configureTransactionState(ProducerConfig config,
LogContext logContext) {

TransactionManager transactionManager = null;

final boolean userConfiguredIdempotence = config.originals().containsKey(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG);
final boolean userConfiguredTransactions = config.originals().containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
if (userConfiguredTransactions && !userConfiguredIdempotence)
log.info("Overriding the default {} to true since {} is specified.", ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,
ProducerConfig.TRANSACTIONAL_ID_CONFIG);

Comment on lines 517 to 522

@showuon showuon Jan 19, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We won't overriding the default ENABLE_IDEMPOTENCE_CONFIG because we already default to true. No need to inform users.

if (config.idempotenceEnabled()) {
final String transactionalId = config.getString(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
final int transactionTimeoutMs = config.getInt(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG);
Expand All @@ -540,28 +534,6 @@ private TransactionManager configureTransactionState(ProducerConfig config,
return transactionManager;
}

private static int configureInflightRequests(ProducerConfig config) {
if (config.idempotenceEnabled() && 5 < config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) {
throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" +
" to use the idempotent producer.");
}
return config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
}

private static short configureAcks(ProducerConfig config, Logger log) {
boolean userConfiguredAcks = config.originals().containsKey(ProducerConfig.ACKS_CONFIG);
short acks = Short.parseShort(config.getString(ProducerConfig.ACKS_CONFIG));

if (config.idempotenceEnabled()) {
if (!userConfiguredAcks)
log.info("Overriding the default {} to all since idempotence is enabled.", ProducerConfig.ACKS_CONFIG);
else if (acks != -1)
throw new ConfigException("Must set " + ProducerConfig.ACKS_CONFIG + " to all in order to use the idempotent " +
"producer. Otherwise we cannot guarantee idempotence.");
}
return acks;
}

Comment on lines 543 to 564

@showuon showuon Jan 19, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move this idempotence configs validation into ProducerConfig#postProcessParsedConfig.

/**
* Needs to be called before any other methods when the transactional.id is set in the configuration.
*
Expand Down
42 changes: 19 additions & 23 deletions clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -438,15 +438,14 @@ public class ProducerConfig extends AbstractConfig {
@Override
protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
Map<String, Object> refinedConfigs = CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
maybeOverrideEnableIdempotence(refinedConfigs);
postProcessAndValidateIdempotenceConfigs(refinedConfigs);
maybeOverrideClientId(refinedConfigs);
maybeOverrideAcksAndRetries(refinedConfigs);
return refinedConfigs;
}

private void maybeOverrideClientId(final Map<String, Object> configs) {
String refinedClientId;
boolean userConfiguredClientId = this.originals().containsKey(CLIENT_ID_CONFIG);
boolean userConfiguredClientId = this.hasKeyInOriginals(CLIENT_ID_CONFIG);
if (userConfiguredClientId) {
refinedClientId = this.getString(CLIENT_ID_CONFIG);
} else {
Expand All @@ -456,33 +455,29 @@ private void maybeOverrideClientId(final Map<String, Object> configs) {
configs.put(CLIENT_ID_CONFIG, refinedClientId);
}

private void maybeOverrideEnableIdempotence(final Map<String, Object> configs) {
boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
boolean userConfiguredTransactions = this.originals().containsKey(TRANSACTIONAL_ID_CONFIG);

if (userConfiguredTransactions && !userConfiguredIdempotence) {
configs.put(ENABLE_IDEMPOTENCE_CONFIG, true);
}
}
Comment on lines 459 to 466

@showuon showuon Jan 19, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

unnecessary method since ENABLE_IDEMPOTENCE_CONFIG is default to true now.


private void maybeOverrideAcksAndRetries(final Map<String, Object> configs) {
private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object> configs) {

@showuon showuon Jan 19, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We won't override any configs now, change the method name.

final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
// For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` might need to be overridden.

// For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
if (idempotenceEnabled()) {
boolean userConfiguredRetries = this.originals().containsKey(RETRIES_CONFIG);
if (this.getInt(RETRIES_CONFIG) == 0) {
boolean userConfiguredRetries = hasKeyInOriginals(RETRIES_CONFIG);
if (userConfiguredRetries && this.getInt(RETRIES_CONFIG) == 0) {
throw new ConfigException("Must set " + ProducerConfig.RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
}
configs.put(RETRIES_CONFIG, userConfiguredRetries ? this.getInt(RETRIES_CONFIG) : Integer.MAX_VALUE);

boolean userConfiguredAcks = this.originals().containsKey(ACKS_CONFIG);
boolean userConfiguredAcks = hasKeyInOriginals(ACKS_CONFIG);
final short acks = Short.valueOf(acksStr);
if (userConfiguredAcks && acks != (short) -1) {
throw new ConfigException("Must set " + ACKS_CONFIG + " to all in order to use the idempotent " +
"producer. Otherwise we cannot guarantee idempotence.");
}
configs.put(ACKS_CONFIG, "-1");

boolean userConfiguredInflightRequests = hasKeyInOriginals(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
if (userConfiguredInflightRequests && 5 < this.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) {

@kirktrue kirktrue Jan 28, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not being familiar with the code, can we make 5 a constant somewhere with a brief explanation as to why 5 is a magic number?

@showuon showuon Jan 29, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good suggestion! Updated.

throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" +
" to use the idempotent producer.");
}

@showuon showuon Jan 19, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

add validation for MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION here

}
}

Expand Down Expand Up @@ -514,11 +509,12 @@ public ProducerConfig(Map<String, Object> props) {
}

boolean idempotenceEnabled() {
boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
boolean userConfiguredTransactions = this.originals().containsKey(TRANSACTIONAL_ID_CONFIG);
boolean idempotenceEnabled = userConfiguredIdempotence && this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);

if (!idempotenceEnabled && userConfiguredIdempotence && userConfiguredTransactions)
boolean userConfiguredIdempotence = this.hasKeyInOriginals(ENABLE_IDEMPOTENCE_CONFIG);
boolean userConfiguredTransactions = this.hasKeyInOriginals(TRANSACTIONAL_ID_CONFIG);
boolean idempotenceEnabled = !userConfiguredIdempotence || this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);

@showuon showuon Jan 19, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the key fix: before the PR, we only considered the idempotenceEnabled when user explicitly set the config and value to true:

 boolean idempotenceEnabled = userConfiguredIdempotence && this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);

Now, change it to fix the issue.

boolean idempotenceEnabled = !userConfiguredIdempotence || this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);

// default value for enable.idempotence is true since v3.0, it must be wrong setting if transactionId set and idempotence disabled
if (!idempotenceEnabled && userConfiguredTransactions)
throw new ConfigException("Cannot set a " + ProducerConfig.TRANSACTIONAL_ID_CONFIG + " without also enabling idempotence.");
return userConfiguredTransactions || idempotenceEnabled;

@hachikuji hachikuji Feb 4, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: I think the check for userConfiguredTransactions is redundant now.

@showuon showuon Feb 5, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right! After we default the enable.idempotence to true, there's no way to have idempotenceEnabled==false but userConfiguredTransactions==true. Remove it.

}
Expand Down
12 changes: 9 additions & 3 deletions clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,15 @@
import org.slf4j.LoggerFactory;
import org.apache.kafka.common.config.provider.ConfigProvider;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.HashSet;
import java.util.HashMap;
import java.util.Objects;
import java.util.TreeMap;
import java.util.ArrayList;

@mimaison mimaison Jan 25, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's keep the current import order

@showuon showuon Jan 26, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Weird, not sure why intellij "help" me reorder it. Updated!

import java.util.concurrent.ConcurrentHashMap;

/**
Expand Down Expand Up @@ -226,6 +227,11 @@ public Set<String> unused() {
return keys;
}

public boolean hasKeyInOriginals(String configKey) {

@mimaison mimaison Jan 25, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What about originalsContainsKey()?

Objects.requireNonNull(configKey, "config key cannot be null");

@mimaison mimaison Jan 25, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need this null check? Shouldn't we rely on the behaviour of the underlying Map?

@showuon showuon Jan 26, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Make sense! We can rely on the underlying map.

return originals.containsKey(configKey);
}

public Map<String, Object> originals() {
Map<String, Object> copy = new RecordingMap<>();
copy.putAll(originals);
Expand Down
Loading