You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/02/17 13:16:41 UTC
[GitHub] [kafka] showuon opened a new pull request #11788: KAFKA-13673: disable idempotence when config conflicts
showuon opened a new pull request #11788:
URL: https://github.com/apache/kafka/pull/11788
JIRA: [KAFKA-13673](https://issues.apache.org/jira/browse/KAFKA-13673)
Disable idempotence when config conflicts and add tests
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma merged pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma merged pull request #11788:
URL: https://github.com/apache/kafka/pull/11788
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] michalxo commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
michalxo commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1054317604
I ran almost full kafka test suite. Got in total around 23 failures, which is great compared to `trunk` where fails almost everything.
I have executed it last Wednesday, so might not have all the changes since then.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r815260093
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +467,53 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
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` need validation
- if (idempotenceEnabled()) {
- 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.");
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
+
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
+ final int retries = this.getInt(RETRIES_CONFIG);
+ if (retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.info("Idempotence will be disabled because {} is set to 0.", RETRIES_CONFIG, retries);
+ shouldDisableIdempotence = true;
}
- 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 " +
+ if (acks != (short) -1) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + ACKS_CONFIG + " to all in order to use the idempotent " +
"producer. Otherwise we cannot guarantee idempotence.");
+ }
+ log.info("Idempotence will be disabled because {} is set to {}, not set to 'all'.", ACKS_CONFIG, acks);
+ shouldDisableIdempotence = true;
}
- 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" +
+ final int inFlightConnection = this.getInt(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
+ if (MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE < inFlightConnection) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" +
" to use the idempotent producer.");
+ }
+ log.warn("Idempotence will be disabled because {} is set to {}, which is greater than 5. " +
+ "Please note that in v4.0.0 and onward, this will become an error.", MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, inFlightConnection);
+ shouldDisableIdempotence = true;
}
}
+
+ if (shouldDisableIdempotence) {
+ configs.put(ENABLE_IDEMPOTENCE_CONFIG, false);
+ }
+
+ // validate `transaction.id` after validating idempotence dependant configs because `enable.idempotence` config might be overridden
+ idempotenceEnabled = idempotenceEnabled && !shouldDisableIdempotence;
Review comment:
Good suggestion! Updated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r816997743
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -202,25 +208,31 @@
/** <code>metric.reporters</code> */
public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG;
+ // 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>max.in.flight.requests.per.connection</code> */
public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection";
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;
+ + " message re-ordering after a failed send due to retries (i.e., if retries are enabled)."
+ + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
Review comment:
Nit: remove "Note". "Additionally, enabling..." reads better.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r810402844
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.warn("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
Thinking about this some more, not sure there's a lot of value in forcing users to set `idempotence=false` in cases where they're setting `acks=1|0` or `retries=0`. So, I'd change the warning to info for these cases.
`max.in.flight.requests.per.connection` is different since it's an implementation constraint that `idempotence` doesn't work with it vs inherent to the configuration. For this one, I'd have a warning and mention that it will become an error in Kafka 4.0.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1047478002
@mimaison , updated the PR in this commit: https://github.com/apache/kafka/pull/11788/commits/29bb358a2e63046ea332981e8a5864caa9b25e78. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
mimaison commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1058058072
Back porting to 3.0 and 3.1 since we merged https://github.com/apache/kafka/commit/e6db0ca48c42c03655e3781b46ac09dae75cf33e into these branches
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r818344273
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -216,21 +216,22 @@
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)."
- + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " Additionally, enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
- + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second"
- + " succeeds, then the records in the second batch may appear first. Note additionally that produce requests will be"
- + " failed before the number of retries has been exhausted if the timeout configured by"
+ + " Produce requests will be failed before the number of retries has been exhausted if the timeout configured by"
+ " <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> expires first before successful acknowledgement. Users should generally"
+ " prefer to leave this config unset and instead use <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> to control"
+ " retry behavior."
- + " <p>"
+ + "<p>"
+ + "Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
I see. Updated. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r810561983
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.warn("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
Sounds good to me. I've updated the PR. Also appended a note to upgrade doc. Please take a look. Thank you.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
mimaison commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r811310081
##########
File path: docs/upgrade.html
##########
@@ -73,7 +74,8 @@ <h5><a id="upgrade_311_notable" href="#upgrade_311_notable">Notable changes in 3
<ul>
<li>A bug prevented the producer idempotence default from being applied which meant that it remained disabled unless the user had explicitly set
<code>enable.idempotence</code> to true. See <a href="https://issues.apache.org/jira/browse/KAFKA-13598">KAFKA-13598</a>for more details.
- This issue was fixed and the default is properly applied.</li>
+ This issue was fixed and the default is properly applied.
+ However, if <code>enable.idempotence</code> is unset and the other configs conflicts with it, idempotence will be disabled to avoid break existing producers.</li>
Review comment:
`break` -> `breaking`
Same below
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,56 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.info("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
I think it would read better if we remove `config` from the sentence.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r809037746
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
##########
@@ -514,7 +514,7 @@ private TransactionManager configureTransactionState(ProducerConfig config,
LogContext logContext) {
TransactionManager transactionManager = null;
- if (config.idempotenceEnabled()) {
+ if (config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)) {
Review comment:
refactor: the `idempotenceEnabled()` method is only to get the config of `ENABLE_IDEMPOTENCE_CONFIG` now.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r813489168
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -276,8 +276,8 @@
+ "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 " + 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.";
+ + ACKS_CONFIG + "</code> must be 'all'. If incompatible values are set, a <code>ConfigException</code> will be thrown. "
+ + "The default value is `true`. But if incompatible values are set and this config is not set explicitly, idempotent producer will be disabled automatically.";
Review comment:
This is a difficult one. The update looks better than my current change. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r812992120
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,56 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.info("`enable.idempotence` will be disabled because {} is set to 0.", RETRIES_CONFIG, retries);
Review comment:
Nit: I think we can simply say `Idempotence will be disabled...` (instead of `enable.idempotence` will be disabled...`)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r809621101
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.warn("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
I agree we directly throw exception after AK 4.0.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r812221236
##########
File path: docs/upgrade.html
##########
@@ -24,7 +24,8 @@ <h5><a id="upgrade_320_notable" href="#upgrade_320_notable">Notable changes in 3
<li>Idempotence for the producer is enabled by default. In 3.0.0 and 3.1.0, a bug prevented this default from being applied,
which meant that idempotence remained disabled unless the user had explicitly set <code>enable.idempotence</code> to true
(See <a href="https://issues.apache.org/jira/browse/KAFKA-13598">KAFKA-13598</a>for more details).
- This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.</li>
+ This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.
+ However, if <code>enable.idempotence</code> is unset and the other configs conflicts with it, idempotence will be disabled to avoid breaking existing producers.</li>
Review comment:
Similar feedback for the other two updates below.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r814763487
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +467,53 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
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` need validation
- if (idempotenceEnabled()) {
- 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.");
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
+
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
+ final int retries = this.getInt(RETRIES_CONFIG);
+ if (retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.info("Idempotence will be disabled because {} is set to 0.", RETRIES_CONFIG, retries);
+ shouldDisableIdempotence = true;
}
- 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 " +
+ if (acks != (short) -1) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + ACKS_CONFIG + " to all in order to use the idempotent " +
"producer. Otherwise we cannot guarantee idempotence.");
+ }
+ log.info("Idempotence will be disabled because {} is set to {}, not set to 'all'.", ACKS_CONFIG, acks);
+ shouldDisableIdempotence = true;
}
- 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" +
+ final int inFlightConnection = this.getInt(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
+ if (MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE < inFlightConnection) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" +
" to use the idempotent producer.");
+ }
+ log.warn("Idempotence will be disabled because {} is set to {}, which is greater than 5. " +
+ "Please note that in v4.0.0 and onward, this will become an error.", MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, inFlightConnection);
+ shouldDisableIdempotence = true;
}
}
+
+ if (shouldDisableIdempotence) {
+ configs.put(ENABLE_IDEMPOTENCE_CONFIG, false);
+ }
+
+ // validate `transaction.id` after validating idempotence dependant configs because `enable.idempotence` config might be overridden
+ idempotenceEnabled = idempotenceEnabled && !shouldDisableIdempotence;
Review comment:
Maybe we can set this to `false` in the `shouldDisableIdempotence` block? Seems a bit more natural.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r812220909
##########
File path: docs/upgrade.html
##########
@@ -24,7 +24,8 @@ <h5><a id="upgrade_320_notable" href="#upgrade_320_notable">Notable changes in 3
<li>Idempotence for the producer is enabled by default. In 3.0.0 and 3.1.0, a bug prevented this default from being applied,
which meant that idempotence remained disabled unless the user had explicitly set <code>enable.idempotence</code> to true
(See <a href="https://issues.apache.org/jira/browse/KAFKA-13598">KAFKA-13598</a>for more details).
- This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.</li>
+ This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.
+ However, if <code>enable.idempotence</code> is unset and the other configs conflicts with it, idempotence will be disabled to avoid breaking existing producers.</li>
Review comment:
Maybe we can update the first sentence. "Idempotence for the producer is enabled by default if no conflicting configurations are set explicitly." I don't think we need the "However" part, we can include that detail in the KIP's compatibility section instead.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r818340881
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -216,21 +216,22 @@
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)."
- + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " Additionally, enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
- + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second"
- + " succeeds, then the records in the second batch may appear first. Note additionally that produce requests will be"
- + " failed before the number of retries has been exhausted if the timeout configured by"
+ + " Produce requests will be failed before the number of retries has been exhausted if the timeout configured by"
+ " <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> expires first before successful acknowledgement. Users should generally"
+ " prefer to leave this config unset and instead use <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> to control"
+ " retry behavior."
- + " <p>"
+ + "<p>"
+ + "Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
I meant to have "Note that enabling idempotence requires this config..." before "Allowing retries...". And break the two parts with a paragraph.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r818342370
##########
File path: config/server.properties
##########
@@ -102,18 +102,19 @@ transaction.state.log.min.isr=1
# from the end of the log.
# The minimum age of a log file to be eligible for deletion due to age
-log.retention.hours=168
+log.retention.ms=10000
# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
-log.segment.bytes=1073741824
+log.segment.bytes=1000
+log.retention.check.interval.ms=10000
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
-log.retention.check.interval.ms=300000
+
Review comment:
Oops! Let me revert 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r809041039
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.warn("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
log warning to users to let them know the `enable.idempotence` is disabled.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r809039418
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -511,15 +542,6 @@ public ProducerConfig(Map<String, Object> props) {
super(CONFIG, props);
}
- boolean idempotenceEnabled() {
- boolean userConfiguredTransactions = this.originals().containsKey(TRANSACTIONAL_ID_CONFIG);
- 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 idempotenceEnabled;
- }
Review comment:
Move the `transaction.id` config validation into validation method, so that we don't need to validate it each time when checking `idempotenceEnabled`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r813491031
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
Review comment:
Nice catch! Those checks are for old configs with default `retries == 0`... etc. Updated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r811623527
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -276,8 +276,8 @@
+ "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 " + 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.";
+ + ACKS_CONFIG + "</code> must be 'all'. If incompatible values are set, a <code>ConfigException</code> will be thrown. "
+ + "The default value is `true`. But if incompatible values are set and this config is not set explicitly, idempotent producer will be disabled automatically.";
Review comment:
`If these values are not explicitly set by the user, suitable values will be chosen.`
-> remove this line because we already default all the values to **suitable values**.
Added:
`The default value is 'true'. But if incompatible values are set and this config is not set explicitly, idempotent producer will be disabled automatically.`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r815260473
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -205,25 +208,31 @@
/** <code>metric.reporters</code> */
public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG;
+ // 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>max.in.flight.requests.per.connection</code> */
public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection";
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;
+ + " message re-ordering after a failed send due to retries (i.e., if retries are enabled)."
+ + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
+ + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
side fix: currently, we said:
```
Allowing retries without setting max.in.flight.requests.per.connection to 1 will potentially change the ordering of records
```
which is not quite right, because we didn't have the idempotence before.
Updated to:
```
Allowing retries and disabling enable.idempotence but without setting max.in.flight.requests.per.connection to 1 will potentially change the ordering of records
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r809370789
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.warn("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
One question is what we want to do with this longer term. Do we want to upgrade this to an error in Apache Kafka 4.0?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r813497832
##########
File path: clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
##########
@@ -361,6 +369,20 @@ public void testInflightRequestsAndIdempotenceForIdempotentProducers() {
config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),
"max.in.flight.requests.per.connection should be overwritten");
+ Properties validProps2 = new Properties() {{
+ putAll(baseProps);
+ setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "10");
Review comment:
Agree! Updated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r812219037
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -276,8 +276,8 @@
+ "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 " + 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.";
+ + ACKS_CONFIG + "</code> must be 'all'. If incompatible values are set, a <code>ConfigException</code> will be thrown. "
+ + "The default value is `true`. But if incompatible values are set and this config is not set explicitly, idempotent producer will be disabled automatically.";
Review comment:
We should update the documentation for the 3 configs mentioned here too. I have a suggestion to try and make this particular one a bit clearer regarding the defaults, will post it a bit later.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r816997743
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -202,25 +208,31 @@
/** <code>metric.reporters</code> */
public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG;
+ // 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>max.in.flight.requests.per.connection</code> */
public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection";
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;
+ + " message re-ordering after a failed send due to retries (i.e., if retries are enabled)."
+ + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
Review comment:
Nit: remove "Note".
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1058063555
Merged to trunk, 3.1 and 3.0.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r812532751
##########
File path: docs/upgrade.html
##########
@@ -24,7 +24,8 @@ <h5><a id="upgrade_320_notable" href="#upgrade_320_notable">Notable changes in 3
<li>Idempotence for the producer is enabled by default. In 3.0.0 and 3.1.0, a bug prevented this default from being applied,
which meant that idempotence remained disabled unless the user had explicitly set <code>enable.idempotence</code> to true
(See <a href="https://issues.apache.org/jira/browse/KAFKA-13598">KAFKA-13598</a>for more details).
- This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.</li>
+ This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.
+ However, if <code>enable.idempotence</code> is unset and the other configs conflicts with it, idempotence will be disabled to avoid breaking existing producers.</li>
Review comment:
Yes, looks better! Updated. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r813495690
##########
File path: docs/upgrade.html
##########
@@ -21,7 +21,7 @@
<h5><a id="upgrade_320_notable" href="#upgrade_320_notable">Notable changes in 3.2.0</a></h5>
<ul>
- <li>Idempotence for the producer is enabled by default. In 3.0.0 and 3.1.0, a bug prevented this default from being applied,
+ <li>Idempotence for the producer is enabled by default if no conflicting configurations are set explicitly. In 3.0.0 and 3.1.0, a bug prevented this default from being applied,
Review comment:
No worries. I agree we should make it as clear as possible, and this update indeed makes it better!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
mimaison commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1055678947
@ijuma Can you take another look? I'm just waiting on this before getting started on 3.0.1.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r816996953
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -205,25 +208,31 @@
/** <code>metric.reporters</code> */
public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG;
+ // 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>max.in.flight.requests.per.connection</code> */
public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection";
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;
+ + " message re-ordering after a failed send due to retries (i.e., if retries are enabled)."
+ + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
+ + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
Maybe we need to reword this a bit given the new defaults. Move the following nearer the top:
> Users should generally"
> + " prefer to leave this config unset and instead use <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> to control
Move the paragraph regarding what happens if idempotence is disabled to the end.
Thoughts?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1055087971
Thanks for verifying it @michalxo !
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1055690966
@mimaison @showuon I left a couple more minor comments that should hopefully be easy to address.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r817287946
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -205,25 +208,31 @@
/** <code>metric.reporters</code> */
public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG;
+ // 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>max.in.flight.requests.per.connection</code> */
public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection";
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;
+ + " message re-ordering after a failed send due to retries (i.e., if retries are enabled)."
+ + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
+ + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
@ijuma Agree! Below is what I've updated. I think it looks better.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1058063862
@mimaison I did the backports.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r818184023
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -216,21 +216,22 @@
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)."
- + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " Additionally, enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
- + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second"
- + " succeeds, then the records in the second batch may appear first. Note additionally that produce requests will be"
- + " failed before the number of retries has been exhausted if the timeout configured by"
+ + " Produce requests will be failed before the number of retries has been exhausted if the timeout configured by"
+ " <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> expires first before successful acknowledgement. Users should generally"
+ " prefer to leave this config unset and instead use <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> to control"
+ " retry behavior."
- + " <p>"
+ + "<p>"
+ + "Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
It reads weird to say
> disabling <code>enable.idempotence</code>
I would write the sentence as:
> Allowing retries while setting <code>enable.idempotence</code> to <code>false</code> and <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
I would also move this paragraph below:
> Enabling idempotence requires this config value to be greater than 0...
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1042944712
@mimaison @hachikuji @vvcephei , call for review. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r809040186
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
Review comment:
throw exception only when user set idempotence to true.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r810402844
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.warn("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
Thinking about this some more, not sure there's a lot of value in forcing users to set `idempotence=false` in cases where they're setting `acks=-1` or `retries=0`. So, I'd change the warning to info for these cases.
`max.in.flight.requests.per.connection` is different since it's an implementation constraint that `idempotence` doesn't work with it vs inherent to the configuration. For this one, I'd have a warning and mention that it will become an error in Kafka 4.0.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r810402844
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.warn("`enable.idempotence` will be disabled because {} config is set to 0.", RETRIES_CONFIG, retries);
Review comment:
Thinking about this some more, not sure there's a lot of value in forcing users to set `idempotence=false` in cases where they're setting `acks=1|0` or `retries=0`. So, I'd change the warning to info for these cases.
`max.in.flight.requests.per.connection` is different since it's an implementation constraint that `idempotence` doesn't work when it's > 5 vs inherent to the configuration. For this one, I'd have a warning and mention that it will become an error in Kafka 4.0.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r811623527
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -276,8 +276,8 @@
+ "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 " + 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.";
+ + ACKS_CONFIG + "</code> must be 'all'. If incompatible values are set, a <code>ConfigException</code> will be thrown. "
+ + "The default value is `true`. But if incompatible values are set and this config is not set explicitly, idempotent producer will be disabled automatically.";
Review comment:
`If these values are not explicitly set by the user, suitable values will be chosen.`
-> remove this line because we already default all the values to **suitable values**.
Added:
`The default value is `true`. But if incompatible values are set and this config is not set explicitly, idempotent producer will be disabled automatically.`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1050477547
@ijuma , please take a look again when available. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r812999106
##########
File path: clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
##########
@@ -361,6 +369,20 @@ public void testInflightRequestsAndIdempotenceForIdempotentProducers() {
config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),
"max.in.flight.requests.per.connection should be overwritten");
+ Properties validProps2 = new Properties() {{
+ putAll(baseProps);
+ setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "10");
Review comment:
Should we use 6 instead of 10?
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,55 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
Review comment:
Hmm, I think this logic and elsewhere is a bit confusing. If `retries == 0` _and_ idempotence has been enabled by the user, we need to throw. It doesn't matter if retries is set by the user or not. Of course, we only expect `retries == 0` if set by the user. But we are hiding a potential bug in the way we're checking this. Same applies for other configs.
##########
File path: docs/upgrade.html
##########
@@ -21,7 +21,7 @@
<h5><a id="upgrade_320_notable" href="#upgrade_320_notable">Notable changes in 3.2.0</a></h5>
<ul>
- <li>Idempotence for the producer is enabled by default. In 3.0.0 and 3.1.0, a bug prevented this default from being applied,
+ <li>Idempotence for the producer is enabled by default if no conflicting configurations are set explicitly. In 3.0.0 and 3.1.0, a bug prevented this default from being applied,
Review comment:
Sorry, after reading this again, I think we can simply say "if no conflicting configurations are set". The "explicitly" feels redundant. Same for the two other paragraphs below. Yes, I know I suggested this text. :)
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -276,8 +276,8 @@
+ "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 " + 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.";
+ + ACKS_CONFIG + "</code> must be 'all'. If incompatible values are set, a <code>ConfigException</code> will be thrown. "
+ + "The default value is `true`. But if incompatible values are set and this config is not set explicitly, idempotent producer will be disabled automatically.";
Review comment:
Can we update `MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC`, `RETRIES_DOC` and `ACKS_DOC` to mention the impact on idempotence?
Also, for this description, I was thinking something like:
> 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 " + 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'.
>
> <p>Idempotence is enabled by default if no conflicting configurations are set. If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled. If idempotence is explicitly enabled and conflicting configurations are set, a ConfigException is thrown.
Thoughts?
##########
File path: clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
##########
@@ -233,6 +233,19 @@ public void testAcksAndIdempotenceForIdempotentProducers() {
config.getString(ProducerConfig.ACKS_CONFIG),
"acks should be overwritten");
+ Properties validProps4 = new Properties() {{
+ putAll(baseProps);
+ setProperty(ProducerConfig.ACKS_CONFIG, "0");
Review comment:
Do we also want a test for acks=1?
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -461,27 +464,56 @@ private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object>
final Map<String, Object> originalConfigs = this.originals();
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
+ final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
+ boolean shouldDisableIdempotence = false;
- // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation
- if (idempotenceEnabled()) {
+ // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
+ if (idempotenceEnabled) {
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.");
+ int retries = this.getInt(RETRIES_CONFIG);
+ if (userConfiguredRetries && retries == 0) {
+ if (userConfiguredIdempotence) {
+ throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
+ }
+ log.info("`enable.idempotence` will be disabled because {} is set to 0.", RETRIES_CONFIG, retries);
Review comment:
Nit: I think we can simply say `Idempotence will be disabled`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r813497899
##########
File path: clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
##########
@@ -233,6 +233,19 @@ public void testAcksAndIdempotenceForIdempotentProducers() {
config.getString(ProducerConfig.ACKS_CONFIG),
"acks should be overwritten");
+ Properties validProps4 = new Properties() {{
+ putAll(baseProps);
+ setProperty(ProducerConfig.ACKS_CONFIG, "0");
Review comment:
New added. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1048388033
@showuon Thanks for the patch. The approach seems reasonable. At least I cannot think of a better alternative. If the user is explicitly opting into weaker semantics, then it seems fair to disable idempotence. The inflight limitation is vexing, but I hope we can do something about that separately.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r818341161
##########
File path: config/server.properties
##########
@@ -102,18 +102,19 @@ transaction.state.log.min.isr=1
# from the end of the log.
# The minimum age of a log file to be eligible for deletion due to age
-log.retention.hours=168
+log.retention.ms=10000
# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
-log.segment.bytes=1073741824
+log.segment.bytes=1000
+log.retention.check.interval.ms=10000
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
-log.retention.check.interval.ms=300000
+
Review comment:
Are these changes accidental?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mimaison commented on pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
mimaison commented on pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#issuecomment-1058064726
Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r818255518
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -216,21 +216,22 @@
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)."
- + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " Additionally, enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
- + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second"
- + " succeeds, then the records in the second batch may appear first. Note additionally that produce requests will be"
- + " failed before the number of retries has been exhausted if the timeout configured by"
+ + " Produce requests will be failed before the number of retries has been exhausted if the timeout configured by"
+ " <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> expires first before successful acknowledgement. Users should generally"
+ " prefer to leave this config unset and instead use <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> to control"
+ " retry behavior."
- + " <p>"
+ + "<p>"
+ + "Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
@ijuma , thanks for the comment. I've updated as below (though I'm not 100% sure I catch your point about this one: `I would also move this paragraph below...`.
<img width="837" alt="image" src="https://user-images.githubusercontent.com/43372967/156482216-a487616b-acf1-4568-bc10-9ba2afea957d.png">
This is my update. I think we can merge this PR first to cut the RC for v3.0.1 first. And we can have follow-up PR if you have any other comments. Thank you for your review!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11788: KAFKA-13673: disable idempotence when config conflicts
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11788:
URL: https://github.com/apache/kafka/pull/11788#discussion_r818345167
##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -216,21 +216,22 @@
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)."
- + " Note additionally that enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ + " Additionally, enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "."
+ " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.";
/** <code>retries</code> */
public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ " Note that this retry is no different than if the client resent the record upon receiving the error."
- + " Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
- + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second"
- + " succeeds, then the records in the second batch may appear first. Note additionally that produce requests will be"
- + " failed before the number of retries has been exhausted if the timeout configured by"
+ + " Produce requests will be failed before the number of retries has been exhausted if the timeout configured by"
+ " <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> expires first before successful acknowledgement. Users should generally"
+ " prefer to leave this config unset and instead use <code>" + DELIVERY_TIMEOUT_MS_CONFIG + "</code> to control"
+ " retry behavior."
- + " <p>"
+ + "<p>"
+ + "Allowing retries and disabling <code>enable.idempotence</code> but without setting <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to 1 will potentially change the"
Review comment:
After the latest update:
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org