showuon commented on a change in pull request #11691:
URL: https://github.com/apache/kafka/pull/11691#discussion_r795016730
##########
File path:
clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
##########
@@ -456,33 +455,29 @@ private void maybeOverrideClientId(final Map<String,
Object> configs) {
configs.put(CLIENT_ID_CONFIG, refinedClientId);
}
- private void maybeOverrideEnableIdempotence(final Map<String, Object>
configs) {
- boolean userConfiguredIdempotence =
this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
- boolean userConfiguredTransactions =
this.originals().containsKey(TRANSACTIONAL_ID_CONFIG);
-
- if (userConfiguredTransactions && !userConfiguredIdempotence) {
- configs.put(ENABLE_IDEMPOTENCE_CONFIG, true);
- }
- }
-
- private void maybeOverrideAcksAndRetries(final Map<String, Object>
configs) {
+ private void postProcessAndValidateIdempotenceConfigs(final Map<String,
Object> configs) {
final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
configs.put(ACKS_CONFIG, acksStr);
- // For idempotence producers, values for `RETRIES_CONFIG` and
`ACKS_CONFIG` might need to be overridden.
+
+ // For idempotence producers, values for `RETRIES_CONFIG` and
`ACKS_CONFIG` need validation
if (idempotenceEnabled()) {
- boolean userConfiguredRetries =
this.originals().containsKey(RETRIES_CONFIG);
- if (this.getInt(RETRIES_CONFIG) == 0) {
+ boolean userConfiguredRetries =
originalsContainsKey(RETRIES_CONFIG);
+ if (userConfiguredRetries && this.getInt(RETRIES_CONFIG) == 0) {
throw new ConfigException("Must set " +
ProducerConfig.RETRIES_CONFIG + " to non-zero when using the idempotent
producer.");
}
- configs.put(RETRIES_CONFIG, userConfiguredRetries ?
this.getInt(RETRIES_CONFIG) : Integer.MAX_VALUE);
- boolean userConfiguredAcks =
this.originals().containsKey(ACKS_CONFIG);
+ boolean userConfiguredAcks = originalsContainsKey(ACKS_CONFIG);
final short acks = Short.valueOf(acksStr);
if (userConfiguredAcks && acks != (short) -1) {
throw new ConfigException("Must set " + ACKS_CONFIG + " to all
in order to use the idempotent " +
"producer. Otherwise we cannot guarantee
idempotence.");
}
- configs.put(ACKS_CONFIG, "-1");
+
+ boolean userConfiguredInflightRequests =
originalsContainsKey(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
+ if (userConfiguredInflightRequests && 5 <
this.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) {
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
