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 16 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
5 changes: 3 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,9 @@ 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)) {
Map<String, Object> originalConfig = config.originals();
if ((!originalConfig.containsKey(RECONNECT_BACKOFF_MAX_MS_CONFIG)) &&
originalConfig.containsKey(RECONNECT_BACKOFF_MS_CONFIG)) {
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
34 changes: 3 additions & 31 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 @@ -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
45 changes: 21 additions & 24 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 @@ -207,6 +207,8 @@ public class ProducerConfig extends AbstractConfig {
private static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC = "The maximum number of unacknowledged requests the client will send on a single connection before blocking."
+ " Note that if this config is set to be greater than 1 and <code>enable.idempotence</code> is set to false, there is a risk of"
+ " message re-ordering after a failed send due to retries (i.e., if retries are enabled).";
// max.in.flight.requests.per.connection should be less than or equal to 5 when idempotence producer enabled to ensure message ordering
private static final int MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE = 5;

/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
Expand Down Expand Up @@ -269,8 +271,8 @@ public class ProducerConfig extends AbstractConfig {
public static final String ENABLE_IDEMPOTENCE_CONFIG = "enable.idempotence";
public static final String ENABLE_IDEMPOTENCE_DOC = "When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream. If 'false', producer "
+ "retries due to broker failures, etc., may write duplicates of the retried message in the stream. "
+ "Note that enabling idempotence requires <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to be less than or equal to 5 "
+ "(with message ordering preserved for any allowable value), <code>" + RETRIES_CONFIG + "</code> to be greater than 0, and <code>"
+ "Note that enabling idempotence requires <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE
+ " (with message ordering preserved for any allowable value), <code>" + RETRIES_CONFIG + "</code> to be greater than 0, and <code>"
+ ACKS_CONFIG + "</code> must be 'all'. If these values are not explicitly set by the user, suitable values will be chosen. If incompatible "
+ "values are set, a <code>ConfigException</code> will be thrown.";

Expand Down Expand Up @@ -438,9 +440,8 @@ 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;
}

Expand All @@ -456,33 +457,30 @@ 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 Map<String, Object> originalConfigs = this.originals();
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 = originalConfigs.containsKey(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 = originalConfigs.containsKey(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 = originalConfigs.containsKey(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
if (userConfiguredInflightRequests && MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE < this.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.");
}
}
}

Expand Down Expand Up @@ -514,12 +512,11 @@ 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 idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
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
Loading