KAFKA-10199: Enable state updater by default (#16107)

We have already enabled the state updater by default once. However, we ran into issues that forced us to disable it again. We think that we fixed those issues. So we want to enable the state updater again by default. Reviewers: Lucas Brutschy <lbrutschy@confluent.io>, Matthias J. Sax <matthias@confluent.io>
apache · Jun 12, 2024 · 39ffdea · 39ffdea
1 parent 0782232
commit 39ffdea
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -1273,7 +1273,7 @@ public static class InternalConfig {
         public static final String STATE_UPDATER_ENABLED = "__state.updater.enabled__";
 
         public static boolean getStateUpdaterEnabled(final Map<String, Object> configs) {
-            return InternalConfig.getBoolean(configs, InternalConfig.STATE_UPDATER_ENABLED, false);
+            return InternalConfig.getBoolean(configs, InternalConfig.STATE_UPDATER_ENABLED, true);
         }
 
         // Private API to enable processing threads (i.e. polling is decoupled from processing)

diff --git a/.../src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/.../src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
@@ -468,10 +468,10 @@ private void shouldPollWithRightTimeout(final Properties properties) {
             assertEquals(Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG)), consumer.lastPollTimeout());
         } else {
             if (!properties.containsKey(InternalConfig.STATE_UPDATER_ENABLED)
-                    || !((boolean) properties.get(InternalConfig.STATE_UPDATER_ENABLED))) {
-                assertEquals(Duration.ZERO, consumer.lastPollTimeout());
-            } else {
+                    || (boolean) properties.get(InternalConfig.STATE_UPDATER_ENABLED)) {
                 assertEquals(Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG)), consumer.lastPollTimeout());
+            } else {
+                assertEquals(Duration.ZERO, consumer.lastPollTimeout());
             }
         }
     }