KAFKA-10199: Enable state updater by default (apache#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 <[email protected]>, Matthias J. Sax <[email protected]>
gongxuanzhang · Jun 12, 2024 · 007f480 · 007f480
1 parent 0f0b516
commit 007f480
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());
             }
         }
     }