You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by bb...@apache.org on 2026/08/04 14:45:14 UTC
(kafka) branch trunk updated: KAFKA-20504: Update state wipe tests (#22959)
This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 5da1831e1f6 KAFKA-20504: Update state wipe tests (#22959)
5da1831e1f6 is described below
commit 5da1831e1f6b27018550302a52b9f59bc9403ce9
Author: Nick Telford <ni...@gmail.com>
AuthorDate: Tue Aug 4 15:45:06 2026 +0100
KAFKA-20504: Update state wipe tests (#22959)
Existing tests verifying state-wipe behaviour on dirty EOS close need to
be scoped to the non-transactional case, since transactional state
stores never let uncommitted writes reach the base store and so should
not be wiped on error.
Most of this was already covered by earlier KIP-892 work —
`StateManagerUtilTest`, `StandbyTaskTest`, `ProcessorStateManagerTest`,
and `StandbyTaskEOSIntegrationTest` already exercise both the
transactional and non-transactional paths correctly. Two gaps remained:
- `StreamTaskTest` had a non-transactional active-task wipe test but no
transactional counterpart (unlike its `StandbyTaskTest` sibling), so
this adds
`shouldNotWipeStateDirectoryWhenCloseDirtyAndEosEnabledWithTransactionalStateStores`
mirroring the existing pattern.
- `EOSUncleanShutdownIntegrationTest` had no transactional coverage at
all, and mutated a shared static `Properties` field in place (a footgun
once a second test method exists). This refactors both tests to build
their own `Properties` copy, makes the existing test explicitly
non-transactional, and adds
`shouldNotWipeStateStoreOnUncleanShutdownWhenTransactional` asserting
the state directory survives an unclean shutdown when transactional
stores are enabled.
Reviewers: Bill Bejeck <bb...@apache.org>
---
.../EOSUncleanShutdownIntegrationTest.java | 87 +++++++++++++++++++++-
.../processor/internals/StreamTaskTest.java | 31 ++++++++
2 files changed, 115 insertions(+), 3 deletions(-)
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EOSUncleanShutdownIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EOSUncleanShutdownIntegrationTest.java
index 32f5dd69b75..e4d1c0c702d 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EOSUncleanShutdownIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/EOSUncleanShutdownIntegrationTest.java
@@ -86,11 +86,16 @@ public class EOSUncleanShutdownIntegrationTest {
private static final int RECORD_TOTAL = 3;
+ // Exercises the non-transactional store path: without transactional state stores, an unclean
+ // shutdown under EOS may leave uncommitted data on disk, so the state directory is wiped.
@Test
public void shouldWorkWithUncleanShutdownWipeOutStateStore() throws InterruptedException {
final String appId = "shouldWorkWithUncleanShutdownWipeOutStateStore";
- STREAMS_CONFIG.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
- STREAMS_CONFIG.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
+ final Properties config = new Properties();
+ config.putAll(STREAMS_CONFIG);
+ config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
+ config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
+ config.put(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG, "false");
final String input = "input-topic";
cleanStateBeforeTest(CLUSTER, input);
@@ -120,7 +125,7 @@ public class EOSUncleanShutdownIntegrationTest {
mkEntry(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ((Serializer<String>) STRING_SERIALIZER).getClass().getName()),
mkEntry(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers())
));
- final KafkaStreams driver = new KafkaStreams(builder.build(), STREAMS_CONFIG);
+ final KafkaStreams driver = new KafkaStreams(builder.build(), config);
driver.cleanUp();
driver.start();
@@ -163,4 +168,80 @@ public class EOSUncleanShutdownIntegrationTest {
quietlyCleanStateAfterTest(CLUSTER, driver);
}
}
+
+ // With transactional state stores enabled, uncommitted writes never reach the base store, so
+ // an unclean shutdown (that is not due to store corruption) must NOT wipe the state directory.
+ @Test
+ public void shouldNotWipeStateStoreOnUncleanShutdownWhenTransactional() throws InterruptedException {
+ final String appId = "shouldNotWipeStateStoreOnUncleanShutdownWhenTransactional";
+ final Properties config = new Properties();
+ config.putAll(STREAMS_CONFIG);
+ config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
+ config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
+ config.put(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG, "true");
+
+ final String input = "input-topic-transactional";
+ cleanStateBeforeTest(CLUSTER, input);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KStream<String, String> inputStream = builder.stream(input);
+
+ final AtomicInteger recordCount = new AtomicInteger(0);
+
+ final KTable<String, String> valueCounts = inputStream
+ .groupByKey()
+ .aggregate(
+ () -> "()",
+ (key, value, aggregate) -> aggregate + ",(" + key + ": " + value + ")",
+ Materialized.as("aggregated_value"));
+
+ valueCounts.toStream().peek((key, value) -> {
+ if (recordCount.incrementAndGet() >= RECORD_TOTAL) {
+ throw new IllegalStateException("Crash on the " + RECORD_TOTAL + " record");
+ }
+ });
+
+ final Properties producerConfig = mkProperties(mkMap(
+ mkEntry(ProducerConfig.CLIENT_ID_CONFIG, "anything"),
+ mkEntry(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ((Serializer<String>) STRING_SERIALIZER).getClass().getName()),
+ mkEntry(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ((Serializer<String>) STRING_SERIALIZER).getClass().getName()),
+ mkEntry(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers())
+ ));
+ final KafkaStreams driver = new KafkaStreams(builder.build(), config);
+ driver.cleanUp();
+ driver.start();
+
+ TestUtils.waitForCondition(() -> driver.state().equals(State.RUNNING),
+ () -> "Expected RUNNING state but driver is on " + driver.state());
+
+ // Task's StateDir
+ final File taskStateDir = new File(String.join("/", TEST_FOLDER.getPath(), appId, "0_0"));
+
+ try {
+ IntegrationTestUtils.produceSynchronously(producerConfig, false, input, Optional.empty(),
+ singletonList(new KeyValueTimestamp<>("k1", "v1", 0L)));
+
+ // wait until the first request is processed and some files are created in it
+ TestUtils.waitForCondition(() -> taskStateDir.exists() && taskStateDir.isDirectory() && taskStateDir.list().length > 0,
+ "Failed awaiting CreateTopics first request failure");
+ IntegrationTestUtils.produceSynchronously(producerConfig, false, input, Optional.empty(),
+ asList(new KeyValueTimestamp<>("k2", "v2", 1L),
+ new KeyValueTimestamp<>("k3", "v3", 2L)));
+
+ TestUtils.waitForCondition(() -> recordCount.get() == RECORD_TOTAL,
+ () -> "Expected " + RECORD_TOTAL + " records processed but only got " + recordCount.get());
+ } catch (final Exception e) {
+ e.printStackTrace();
+ } finally {
+ TestUtils.waitForCondition(() -> driver.state().equals(State.ERROR),
+ () -> "Expected ERROR state but driver is on " + driver.state());
+
+ driver.close();
+
+ assertTrue(taskStateDir.exists());
+
+ quietlyCleanStateAfterTest(CLUSTER, driver);
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index 48639fad16a..eeffde86b29 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -397,6 +397,8 @@ public class StreamTaskTest {
verify(stateDirectory, never()).lock(any());
}
+ // Covers the non-transactional case: without transactional state stores, a dirty close under
+ // EOS may have left uncommitted data on disk, so the state directory must be wiped.
@Test
public void shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() {
when(stateManager.taskId()).thenReturn(taskId);
@@ -426,6 +428,35 @@ public class StreamTaskTest {
inOrder.verify(stateDirectory).unlock(taskId);
}
+ @Test
+ public void shouldNotWipeStateDirectoryWhenCloseDirtyAndEosEnabledWithTransactionalStateStores() {
+ when(stateManager.taskId()).thenReturn(taskId);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ when(stateManager.hasCorruptedStores()).thenReturn(false);
+ // Clean up state directory created as part of setup
+ stateDirectory.close();
+ stateDirectory = mock(StateDirectory.class);
+
+ when(stateDirectory.lock(taskId)).thenReturn(true);
+
+ final StreamsConfig config = createConfig(
+ StreamsConfig.EXACTLY_ONCE_V2,
+ "100",
+ LogAndFailExceptionHandler.class,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class,
+ true);
+
+ task = createStatefulTask(config, true, stateManager);
+ task.suspend();
+ task.closeDirty();
+ task = null;
+
+ // With transactional state stores, the state dir should NOT be wiped on dirty close
+ // unless stores are specifically marked as corrupted.
+ verify(stateManager, never()).baseDir();
+ }
+
@Test
public void shouldResetOffsetsToLastCommittedForSpecifiedPartitions() {
when(stateManager.taskId()).thenReturn(taskId);