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 <[email protected]>
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 <[email protected]>
---
.../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);
