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

showuon · 2022-01-19T07:08:54Z

In short, after v3.0, the default producer is still not idempotence enabled.

In v3.0, we make enable.idempotence default to true, but we didn't adjust the validator and the idempotence enabled check method, so that if user didn't explicitly enable idempotence, this feature won't be turned on. In this PR, I did:

fix the ProducerConfig#idempotenceEnabled method, to make it correctly detect if idempotence is enabled or not
remove some unnecessary config overridden and checks due to we already default acks, retries and enable.idempotence configs.
move the config validator for idempotence producer from KafkaProducer into ProducerConfig. The config validation should be the responsibility of ProducerConfig class.
add an AbstractConfig#hasKeyInOriginals method, to avoid originals configs get copied and only want to check the existence of the key.
fix many broken tests. As mentioned, we didn't actually enable idempotence in v3.0. After this PR, there are some tests broken due to some different behavior between idempotence and non-idempotence producer. Fix them
add tests

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

showuon · 2022-01-19T07:23:02Z

clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java

-        if ((!config.originals().containsKey(RECONNECT_BACKOFF_MAX_MS_CONFIG)) &&
-                config.originals().containsKey(RECONNECT_BACKOFF_MS_CONFIG)) {
+        if ((!config.hasKeyInOriginals(RECONNECT_BACKOFF_MAX_MS_CONFIG)) &&
+                config.hasKeyInOriginals(RECONNECT_BACKOFF_MS_CONFIG)) {


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

showuon · 2022-01-19T07:24:38Z

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

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


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

showuon · 2022-01-19T07:26:23Z

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

-    private static int configureInflightRequests(ProducerConfig config) {
-        if (config.idempotenceEnabled() && 5 < config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) {
-            throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" +
-                    " to use the idempotent producer.");
-        }
-        return config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
-    }
-
-    private static short configureAcks(ProducerConfig config, Logger log) {
-        boolean userConfiguredAcks = config.originals().containsKey(ProducerConfig.ACKS_CONFIG);
-        short acks = Short.parseShort(config.getString(ProducerConfig.ACKS_CONFIG));
-
-        if (config.idempotenceEnabled()) {
-            if (!userConfiguredAcks)
-                log.info("Overriding the default {} to all since idempotence is enabled.", ProducerConfig.ACKS_CONFIG);
-            else if (acks != -1)
-                throw new ConfigException("Must set " + ProducerConfig.ACKS_CONFIG + " to all in order to use the idempotent " +
-                        "producer. Otherwise we cannot guarantee idempotence.");
-        }
-        return acks;
-    }
-


Move this idempotence configs validation into ProducerConfig#postProcessParsedConfig.

showuon · 2022-01-19T07:27:27Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

-    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);
-        }
-    }


unnecessary method since ENABLE_IDEMPOTENCE_CONFIG is default to true now.

showuon · 2022-01-19T07:27:53Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

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


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

showuon · 2022-01-19T07:29:19Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

+            boolean userConfiguredInflightRequests = hasKeyInOriginals(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
+            if (userConfiguredInflightRequests && 5 < this.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) {
+                throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" +
+                        " to use the idempotent producer.");
+            }


add validation for MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION here

showuon · 2022-01-19T07:32:18Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

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

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


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

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

Now, change it to fix the issue.

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

showuon · 2022-01-22T02:21:27Z

@hachikuji @vvcephei , @ijuma , please take a look. Thanks.

mimaison

Thanks @showuon for this nice cleanup!
I left a few questions and suggestions

mimaison · 2022-01-25T15:40:19Z

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

@@ -226,6 +227,11 @@ public Password getPassword(String key) {
        return keys;
    }

+    public boolean hasKeyInOriginals(String configKey) {


What about originalsContainsKey()?

mimaison · 2022-01-25T15:41:32Z

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

@@ -226,6 +227,11 @@ public Password getPassword(String key) {
        return keys;
    }

+    public boolean hasKeyInOriginals(String configKey) {
+        Objects.requireNonNull(configKey, "config key cannot be null");


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

Make sense! We can rely on the underlying map.

mimaison · 2022-01-25T15:42:17Z

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

 import java.util.TreeMap;
+import java.util.ArrayList;


Let's keep the current import order

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

mimaison · 2022-01-25T15:53:42Z

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

@@ -240,6 +242,7 @@ public void testAcksAndIdempotenceForIdempotentProducers() {
        Properties invalidProps2 = new Properties() {{
                putAll(baseProps);
                setProperty(ProducerConfig.ACKS_CONFIG, "1");
+                // explicitly enable idempotence should still throw exception


enable -> enabling

mimaison · 2022-01-25T16:05:55Z

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

        ProducerMetadata metadata = mock(ProducerMetadata.class);

        // Return empty cluster 4 times and cluster from then on
-        when(metadata.fetch()).thenReturn(emptyCluster, emptyCluster, emptyCluster, emptyCluster, onePartitionCluster);
+        if (isIdempotenceEnabled) {


Should we change the comment above?

mimaison · 2022-01-25T16:20:02Z

connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java

@@ -170,6 +170,7 @@ public void configure(final WorkerConfig config) {
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        producerProps.put(ProducerConfig.RETRIES_CONFIG, 0); // we handle retries in this class
+        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false); // disable idempotence since retries is force to 0


Ah :( I assumed Connect was benefiting from idempotency automatically, it looks like it isn't the case

Yes, it might be another PR (or KIP) to enable it in Connect.

mimaison · 2022-01-25T16:20:23Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -2383,6 +2384,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest {

  private def buildTransactionalProducer(): KafkaProducer[Array[Byte], Array[Byte]] = {
    producerConfig.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId)
+    producerConfig.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")


Do we need to explicitly set it?

Yes, because at the beginning of the AuthorizerIntegrationTest test suite, the idempotence is disabled.

You're right, I missed that!

mimaison · 2022-01-25T16:21:12Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -143,6 +143,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
  val adminClients = Buffer[Admin]()

  producerConfig.setProperty(ProducerConfig.ACKS_CONFIG, "1")
+  producerConfig.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false")


Could we instead drop acks=1? I don't think it is important in this test.

Did you miss this comment?

Sorry, I forgot to reply this comment. No we should not remove the acks=1 config to enable the idempotence producer here. In this AuthorizerIntegrationTest test suite, we tested normal producer (created here), and IdempotentProducer (created in buildIdempotentProducer method), and transactional producer (created in buildTransactionalProducer method), so if we remove acks=1 here, we won't be able to test normal producer cases.

mimaison · 2022-01-25T16:40:36Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

@@ -1674,6 +1674,8 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup
      producerProps.put(ProducerConfig.RETRIES_CONFIG, _retries.toString)
      producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, _deliveryTimeoutMs.toString)
      producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, _requestTimeoutMs.toString)
+      // disable the idempotence since some tests want to test the cases when retries=0, and these tests are not testing producers
+      producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false.toString)


Let's use "false"

mimaison · 2022-01-25T16:55:36Z

core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala

    setReadAndWriteAcls(tp2)
-    sendRecords(producer, numRecords, tp2)
+    // in idempotence producer, we need to create another producer because the previous one is in FATEL_ERROR state (due to authorization error)


FATEL_ERROR -> FATAL_ERROR
Do we really expect the idempotent producer to not be usable anymore?

I think that's how it is designed. I added more comment for it to make it clear:
// // in idempotence producer, we need to create another producer because the previous one is in FATEL_ERROR state (due to authorization error)
// If the transaction state in FATEL_ERROR, it'll never transit to other state. check TransactionManager#isTransitionValid for detail

Thank you.

Let's still fix the typo, it should be FATAL_ERROR

ah, you're right, I forgot it.

showuon · 2022-01-27T13:31:51Z

I found there are flaky tests introduced by this PR. I'll fix them first. Thanks.

showuon · 2022-01-28T08:39:30Z

@mimaison , I've updated the PR and fix the flaky tests. The tests will fail because we have 2 threads: Producer thread and Sender thread in producer. And before the latest commit, I have a bad assumption that the sender thread will run after producer send, so the called times verification will be passed. (ex: verify(metadata, times(5)).fetch();)

I fixed it by removing the the need to fetch metadata in sender thread, so that we can focus on testing the producer behavior as we expected.
Thanks.

ijuma · 2022-01-28T16:30:50Z

Are you saying that the change to enable idempotence by default in 3.0 was broken? And we shipped 3.1 with the same critical bug?

hachikuji

Thanks @showuon for the patch. Left a couple comments.

hachikuji · 2022-01-28T17:54:53Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java


-        if (!idempotenceEnabled && userConfiguredIdempotence && userConfiguredTransactions)
+        boolean userConfiguredIdempotence = this.originalsContainsKey(ENABLE_IDEMPOTENCE_CONFIG);


I think this can be simplified a little bit more by getting rid of this variable. How about this?

boolean userConfiguredTransactions = this.originalsContainsKey(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;

Good suggestion! I was affected by the original implementation. Updated.

hachikuji · 2022-01-28T17:58:16Z

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) {


AbstractConfig is sadly a public class. Can we do the PR without this?

Oops! OK, reverted.

hachikuji · 2022-01-28T18:05:30Z

Are you saying that the change to enable idempotence by default in 3.0 was broken? And we shipped 3.1 with the same critical bug?

@ijuma Sadly so. I think the configuration would even have displayed enable.idempotence=true when printed, but the check disables it unless the user had explicitly provided it.

mimaison · 2022-02-16T11:42:19Z

With this change, if a user explicitly sets acks to 0 or 1, we're throwing an error. Currently in order to change acks, users also have to set enable.idempotence=false. I'm expecting many existing applications are currently overriding acks.

So I wonder if instead we should disable idempotency when acks is overridden.

showuon · 2022-02-16T13:06:55Z

@mimaison , do you mean:
So, do you mean, we have some different cases:

if user doesn't set enable.idempotence explicitly: the idempotent producer is only enabled when the acks=all, otherwise, disable idempotence
ex: a. enable.idempotence unset && acks unset => enable idempotence (this was the intent behind the 3.0 change I think)
b. enable.idempotence unset && acks=all => enable idempotence (obviously)
c. enable.idempotence unset && acks!=all => disable idempotence (the case we tried to protect)
if user enable/disable enable.idempotence explicitly, we follow current config validation to throw exceptions when bad config set,
ex: a. enable.idempotence=true && acks=all => enable idempotence (obviously)
b. enable.idempotence=false => disable idempotence
c. enable.idempotence=true && acks!=all => throw exception

And we only do some special check for acks because many users would change this config. For retries and max.in.flight.requests.per.connection, we still follow current config validation no matter enable.idempotence is set explicitly or not.

Does this what you mean?
If so, I'm +1. Actually, if the retries and max.in.flight.requests.per.connection also have the same special check, I also think it makes sense. The purpose is try not to break existing producers after upgrade, which is good.

mimaison · 2022-02-16T14:20:16Z

I too think it makes sense to apply the same logic to retries and max.in.flight.requests.per.connection. When disabling idempotency due to a conflicting config, we should log something.

If we decide to make this change, we should also update the docs of these configs to mention this behavior.

showuon · 2022-02-17T01:43:08Z

@ijuma @hachikuji @vvcephei @kirktrue , any thoughts about this suggestion? This way, we don't need #11767 #11769 to do any change.

vvcephei · 2022-02-17T03:48:25Z

I have to admit that I find all these breakages that are cropping up to be a bit disturbing. I like the idea to default to stronger guarantees, but if the cost is that a lot of existing applications just crash on upgrade, it's not a great tradeoff.

The above logic is fairly complex, but it seems warranted in this case, so I'm +1 as well.

showuon · 2022-02-17T07:12:30Z

KAFKA-13673 is created for this proposal.

ijuma · 2022-02-17T18:56:35Z

I agree with the suggestions here. Especially since the idempotence change did not land in 3.0.0 as originally intended. We can continue the discussion in #11788.

…configs (apache#11691) In v3.0, we changed the default value for `enable.idempotence` to true, but we didn't adjust the validator and the `idempotence` enabled check method. So if a user didn't explicitly enable idempotence, this feature won't be turned on. This patch addresses the problem, cleans up associated logic, and fixes tests that broke as a result of properly applying the new default. Specifically it does the following: 1. fix the `ProducerConfig#idempotenceEnabled` method, to make it correctly detect if `idempotence` is enabled or not 2. remove some unnecessary config overridden and checks due to we already default `acks`, `retries` and `enable.idempotence` configs. 3. move the config validator for the idempotent producer from `KafkaProducer` into `ProducerConfig`. The config validation should be the responsibility of `ProducerConfig` class. 4. add an `AbstractConfig#hasKeyInOriginals` method, to avoid `originals` configs get copied and only want to check the existence of the key. 5. fix many broken tests. As mentioned, we didn't actually enable idempotence in v3.0. After this PR, there are some tests broken due to some different behavior between idempotent and non-idempotent producer. 6. add additional tests to validate configuration behavior Reviewers: Kirk True <kirk@mustardgrain.com>, Ismael Juma <ismael@juma.me.uk>, Mickael Maison <mimaison@users.noreply.github.com>, Jason Gustafson <jason@confluent.io>

Remove 3.2 notes which were accidentally included after cherry-picking apache#11691. Add the section about the idempotence bug in 3.1 notable items. Reviewers: Luke Chen <showuon@gmail.com>, Jason Gustafson <jason@confluent.io>

rondagostino · 2022-03-21T22:29:46Z

This PR broke this system test:

TC_PATHS="tests/kafkatest/tests/tools/log4j_appender_test.py" bash tests/docker/run_tests.sh

showuon · 2022-03-22T07:18:43Z

@rondagostino , yes, we are aware of it, and that's why we have #11788 to disable idempotent producer when there are config conflicts. Could you please run again with the build including #11788 change? Thanks.

rondagostino · 2022-03-22T12:52:14Z

@showuon It still fails at that commit.
git checkout 7c280c1d5f6267acbbebd10d3e58ea4b8fe7a4ef
HEAD is now at 7c280c1d5f KAFKA-13673: disable idempotence when config conflicts (#11788)
./gradlew clean
_DUCKTAPE_OPTIONS="--debug" TC_PATHS="tests/kafkatest/tests/tools/log4j_appender_test.py" bash tests/docker/run_tests.sh
Test fails

showuon · 2022-03-22T13:04:56Z

@rondagostino , could I get the error logs? Thanks.

rondagostino · 2022-03-22T13:50:33Z

Here is kafka_log4j_appender.log.

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/kafka-dev/core/build/dependant-libs-2.13.6/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/tools/build/dependant-libs-2.13.6/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/trogdor/build/dependant-libs-2.13.6/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/api/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/transforms/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/runtime/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/file/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/mirror/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/mirror-client/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/json/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/connect/basic-auth-extension/build/dependant-libs/slf4j-log4j12-1.7.32.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
log4j:WARN No appenders could be found for logger (org.apache.kafka.clients.producer.ProducerConfig).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

Note this line, which seems suspicious (though I am unfamiliar with this tools/what it does/how it does it, so maybe it is irrelevant).

log4j:WARN No appenders could be found for logger (org.apache.kafka.clients.producer.ProducerConfig).

rondagostino · 2022-03-22T14:50:38Z

Test passes with this change, but I don't understand why.
git diff log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java

diff --git a/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java b/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java
index 23272a2cb5..edaf6589a7 100644
--- a/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java
+++ b/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java
@@ -43,6 +43,7 @@ import static org.apache.kafka.clients.producer.ProducerConfig.LINGER_MS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.MAX_BLOCK_MS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.RETRIES_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_JAAS_CONFIG;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_KERBEROS_SERVICE_NAME;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_MECHANISM;
@@ -332,6 +333,7 @@ public class KafkaLog4jAppender extends AppenderSkeleton {

         props.put(KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
         props.put(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+        props.put(ENABLE_IDEMPOTENCE_CONFIG, false);
         this.producer = getKafkaProducer(props);
         LogLog.debug("Kafka producer connected to " + brokerList);
         LogLog.debug("Logging for topic: " + topic);

michalxo · 2022-03-23T07:28:35Z

@showuon (or Yang Yu) found the issue (is it you @showuon ?)
https://issues.apache.org/jira/browse/KAFKA-13761

showuon · 2022-03-23T07:51:35Z

No, I'm not Yang Yu :)
Glad we found out the root cause!

…ate the configs (apache#11691)" This reverts commit 8093331.

…ate the configs (apache#11691)" This reverts commit 43cbf17.

…ate the configs (apache#11691)" This reverts commit 8093331.

…configs (apache#11691) In v3.0, we changed the default value for `enable.idempotence` to true, but we didn't adjust the validator and the `idempotence` enabled check method. So if a user didn't explicitly enable idempotence, this feature won't be turned on. This patch addresses the problem, cleans up associated logic, and fixes tests that broke as a result of properly applying the new default. Specifically it does the following: 1. fix the `ProducerConfig#idempotenceEnabled` method, to make it correctly detect if `idempotence` is enabled or not 2. remove some unnecessary config overridden and checks due to we already default `acks`, `retries` and `enable.idempotence` configs. 3. move the config validator for the idempotent producer from `KafkaProducer` into `ProducerConfig`. The config validation should be the responsibility of `ProducerConfig` class. 4. add an `AbstractConfig#hasKeyInOriginals` method, to avoid `originals` configs get copied and only want to check the existence of the key. 5. fix many broken tests. As mentioned, we didn't actually enable idempotence in v3.0. After this PR, there are some tests broken due to some different behavior between idempotent and non-idempotent producer. 6. add additional tests to validate configuration behavior Reviewers: Kirk True <kirk@mustardgrain.com>, Ismael Juma <ismael@juma.me.uk>, Mickael Maison <mimaison@users.noreply.github.com>, Jason Gustafson <jason@confluent.io>

Remove 3.2 notes which were accidentally included after cherry-picking apache#11691. Add the section about the idempotence bug in 3.1 notable items. Reviewers: Luke Chen <showuon@gmail.com>, Jason Gustafson <jason@confluent.io>

…nd validate the configs (apache#11691)"" This reverts commit dcc96c2.

KAFKA-13598: validate producer config for idempotence enabled

70b240b

showuon commented Jan 19, 2022

View reviewed changes

KAFKA-13598: refactor config.originals

a240180

showuon force-pushed the KAFKA-13598 branch from 6a2d741 to a240180 Compare January 19, 2022 07:39

showuon added 4 commits January 20, 2022 13:19

KAFKA-13598: fix broken tests

da25d7b

KAFKA-13598: fix broken tests

8779dcf

KAFKA-13598: refactor

a5f23b8

KAFKA-13598: fix long hanging test

ca93d15

showuon force-pushed the KAFKA-13598 branch from 843fd82 to ca93d15 Compare January 21, 2022 03:07

showuon added 2 commits January 21, 2022 15:39

Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-13598

14294aa

KAFKA-13598: test

4d5d527

showuon changed the title ~~KAFKA-13598: enable and validate config for idempotence producer~~ KAFKA-13598: enable idempotence producer by default and validate the configs Jan 22, 2022

mimaison reviewed Jan 25, 2022

View reviewed changes

KAFKA-13598: address reviewer's comments to refactor the code

3c2f1de

KAFKA-13598: fix flaky tests and typo

a3c7bcf

showuon force-pushed the KAFKA-13598 branch from b543878 to a3c7bcf Compare January 28, 2022 06:59

showuon added 2 commits January 28, 2022 15:21

Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-13598

03cd033

KAFKA-13598: fix compile error

eadc897

hachikuji reviewed Jan 28, 2022

View reviewed changes

rajinisivaram mentioned this pull request Mar 3, 2022

CONFLUENT: Cherry pick idempotent-by-default fixes from Apache Kafka confluentinc/kafka#671

Merged

3 tasks

LiamClarkeNZ mentioned this pull request Mar 4, 2022

KAFKA-7077: Use default producer settings in Connect Worker #11475

Merged

3 tasks

jeffkbkim added a commit to confluentinc/kafka that referenced this pull request Mar 30, 2022

Revert "KAFKA-13598: enable idempotence producer by default and valid…

bb660d0

…ate the configs (apache#11691)" This reverts commit 8093331.

jeffkbkim added a commit to confluentinc/kafka that referenced this pull request Mar 30, 2022

Revert "KAFKA-13598: enable idempotence producer by default and valid…

3466376

…ate the configs (apache#11691)" This reverts commit 43cbf17.

jeffkbkim added a commit to confluentinc/kafka that referenced this pull request Mar 31, 2022

Revert "KAFKA-13598: enable idempotence producer by default and valid…

f96b021

…ate the configs (apache#11691)" This reverts commit 43cbf17.

jeffkbkim added a commit to confluentinc/kafka that referenced this pull request Mar 31, 2022

Revert "KAFKA-13598: enable idempotence producer by default and valid…

dcc96c2

…ate the configs (apache#11691)" This reverts commit 8093331.

jeffkbkim added a commit to confluentinc/kafka that referenced this pull request Nov 3, 2022

Revert "Revert "KAFKA-13598: enable idempotence producer by default a…

dd42601

…nd validate the configs (apache#11691)"" This reverts commit dcc96c2.

jeffkbkim added a commit to confluentinc/kafka that referenced this pull request Nov 7, 2022

Revert "Revert "KAFKA-13598: enable idempotence producer by default a…

c076dd9

…nd validate the configs (apache#11691)"" This reverts commit dcc96c2.


		if (!idempotenceEnabled && userConfiguredIdempotence && userConfiguredTransactions)
		boolean userConfiguredIdempotence = this.originalsContainsKey(ENABLE_IDEMPOTENCE_CONFIG);

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

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

Conversation

showuon commented Jan 19, 2022 • edited Loading

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

showuon commented Jan 22, 2022

mimaison left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

showuon Jan 27, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

showuon commented Jan 27, 2022

showuon commented Jan 28, 2022

ijuma commented Jan 28, 2022

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji commented Jan 28, 2022 • edited Loading

mimaison commented Feb 16, 2022

showuon commented Feb 16, 2022 • edited Loading

mimaison commented Feb 16, 2022

showuon commented Feb 17, 2022

vvcephei commented Feb 17, 2022

showuon commented Feb 17, 2022

ijuma commented Feb 17, 2022

rondagostino commented Mar 21, 2022

showuon commented Mar 22, 2022

rondagostino commented Mar 22, 2022

showuon commented Mar 22, 2022

rondagostino commented Mar 22, 2022

rondagostino commented Mar 22, 2022

michalxo commented Mar 23, 2022

showuon commented Mar 23, 2022

showuon commented Jan 19, 2022 •

edited

Loading

showuon Jan 27, 2022 •

edited

Loading

hachikuji commented Jan 28, 2022 •

edited

Loading

showuon commented Feb 16, 2022 •

edited

Loading