kirktrue commented on a change in pull request #11691:
URL: https://github.com/apache/kafka/pull/11691#discussion_r794745674
##########
File path:
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
##########
@@ -26,12 +26,13 @@
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+
Review comment:
Super nitpicky -- can we remove this superfluous newline?
##########
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:
Not being familiar with the code, can we make `5` a constant somewhere
with a brief explanation as to why 5 is a magic number?
##########
File path:
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
##########
@@ -26,12 +26,13 @@
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
+import java.util.HashMap;
Review comment:
Super nitpicky -- I think the `import` list is alphabetical, and this is
out of order now.
##########
File path:
clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
##########
@@ -226,6 +227,10 @@ public Password getPassword(String key) {
return keys;
}
+ public boolean originalsContainsKey(String configKey) {
Review comment:
What's the functional difference between
`config.originalsContainsKey("foo")` and
`config.originals().containsKey("foo")`?
I understand the latter creates an unnecessary copy, but beyond that, does
the logic differ?
--
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
