Skip to content

Commit

Permalink
KAFKA-10199: Enable state updater by default (#16107)
Browse files Browse the repository at this point in the history
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>
  • Loading branch information
cadonna committed Jun 12, 2024
1 parent 77a6fe9 commit 8de153e
Showing 2 changed files with 4 additions and 4 deletions.
Original file line number Diff line number Diff line change
@@ -1261,7 +1261,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)
Original file line number Diff line number Diff line change
@@ -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());
}
}
}

0 comments on commit 8de153e

Please sign in to comment.