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 433e99a218d KAFKA-20760: Parameterise PositionRestartIntegrationTest
over transactional mode (#22750)
433e99a218d is described below
commit 433e99a218ddc00f13e008ed6f9a3f166a8842e9
Author: Nick Telford <[email protected]>
AuthorDate: Fri Jul 3 22:08:16 2026 +0100
KAFKA-20760: Parameterise PositionRestartIntegrationTest over transactional
mode (#22750)
Extends `PositionRestartIntegrationTest` to run every store type and
configuration with transactional state stores both enabled and disabled.
This exercises interactive-query position consistency and, crucially,
position recovery on restart for transactional stores — the end-to-end
coverage for the series of transactional state-store position fixes
(KAFKA-20760).
Also adds unit tests for `InMemoryWindowStore` and
`InMemorySessionStore`, mirroring the existing `InMemoryKeyValueStore`
test: each verifies that a transactional store exposes its staged
(uncommitted) position via a READ_UNCOMMITTED query and via
`getPosition()` before commit, and its committed position after commit.
Reviewers: Bill Bejeck <[email protected]>
Co-authored-by: Bill Bejeck <[email protected]>
---
.../PositionRestartIntegrationTest.java | 28 +++++++------
.../state/internals/InMemorySessionStoreTest.java | 48 ++++++++++++++++++++++
.../state/internals/InMemoryWindowStoreTest.java | 40 ++++++++++++++++++
3 files changed, 104 insertions(+), 12 deletions(-)
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
index 5217f133d20..588cdeba05e 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
@@ -255,12 +255,14 @@ public class PositionRestartIntegrationTest {
final List<Arguments> values = new ArrayList<>();
for (final boolean cacheEnabled : Arrays.asList(true, false)) {
for (final boolean logEnabled : Arrays.asList(true, false)) {
- for (final StoresToTest toTest : StoresToTest.values()) {
- // We don't need to test if non-persistent stores without
logging
- // survive restarts, since those are by definition not
durable.
- if (logEnabled || toTest.supplier().get().persistent()) {
- for (final String kind : Arrays.asList("DSL", "PAPI"))
{
- values.add(Arguments.of(cacheEnabled, logEnabled,
toTest, kind));
+ for (final boolean transactional : Arrays.asList(true, false))
{
+ for (final StoresToTest toTest : StoresToTest.values()) {
+ // We don't need to test if non-persistent stores
without logging
+ // survive restarts, since those are by definition not
durable.
+ if (logEnabled ||
toTest.supplier().get().persistent()) {
+ for (final String kind : Arrays.asList("DSL",
"PAPI")) {
+ values.add(Arguments.of(cacheEnabled,
logEnabled, transactional, toTest, kind));
+ }
}
}
}
@@ -361,8 +363,8 @@ public class PositionRestartIntegrationTest {
@ParameterizedTest
@MethodSource("data")
- public void verifyStore(final boolean cache, final boolean log, final
StoresToTest storeToTest, final String kind) {
- final Properties streamsConfig = streamsConfiguration(cache, log,
storeToTest.name(), kind);
+ public void verifyStore(final boolean cache, final boolean log, final
boolean transactional, final StoresToTest storeToTest, final String kind) {
+ final Properties streamsConfig = streamsConfiguration(cache, log,
transactional, storeToTest.name(), kind);
final StreamsBuilder streamsBuilder = getStreamBuilder(cache, log,
storeToTest, kind);
kafkaStreams = IntegrationTestUtils.getStartedStreams(streamsConfig,
streamsBuilder, true);
@@ -647,11 +649,12 @@ public class PositionRestartIntegrationTest {
}
private static Properties streamsConfiguration(final boolean cache,
- final boolean log,
- final String supplier,
- final String kind) {
+ final boolean log,
+ final boolean transactional,
+ final String supplier,
+ final String kind) {
final String safeTestName =
- PositionRestartIntegrationTest.class.getName() + "-" + cache + "-"
+ log + "-"
+ PositionRestartIntegrationTest.class.getName() + "-" + cache + "-"
+ log + "-" + transactional + "-"
+ supplier + "-" + kind;
final Properties config = new Properties();
config.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG,
StreamsConfig.OPTIMIZE);
@@ -668,6 +671,7 @@ public class PositionRestartIntegrationTest {
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
config.put(InternalConfig.IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED, true);
+ config.put(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
transactional);
return config;
}
}
\ No newline at end of file
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
index cfcf156253c..30543189977 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
@@ -17,11 +17,18 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.SessionWindow;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.query.Position;
+import org.apache.kafka.streams.query.PositionBound;
+import org.apache.kafka.streams.query.QueryConfig;
+import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.streams.query.WindowRangeQuery;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.ReadOnlySessionStore;
import org.apache.kafka.test.InternalMockProcessorContext;
@@ -37,6 +44,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.test.StreamsTestUtils.valuesToSet;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -254,6 +263,45 @@ public class InMemorySessionStoreTest extends
AbstractSessionBytesStoreTest {
}
}
+ @Test
+ public void shouldReportUncommittedPositionForTransactionalStore() {
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ props.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(props)
+ );
+ final InMemorySessionStore store = new InMemorySessionStore(
+ "txn-pos-session-store", RETENTION_PERIOD, "scope");
+ store.init(ctx, store);
+ try {
+ ctx.setRecordContext(new ProcessorRecordContext(0, 1, 0, "topic",
new RecordHeaders()));
+ store.put(new Windowed<>(Bytes.wrap("k".getBytes()), new
SessionWindow(0, 10)), "v".getBytes());
+
+ final Position expected = Position.fromMap(mkMap(mkEntry("topic",
mkMap(mkEntry(0, 1L)))));
+
+ // READ_UNCOMMITTED query should expose the staged position before
commit
+ final QueryResult<?> uncommitted = store.query(
+ WindowRangeQuery.withKey(Bytes.wrap("k".getBytes())),
+ PositionBound.unbounded(),
+ new QueryConfig(false));
+ assertEquals(expected, uncommitted.getPosition(),
"READ_UNCOMMITTED query position");
+
+ // getPosition() reports the uncommitted (committed + staged)
position, mirroring
+ // RocksDBStore, so the changelog consistency vector reflects the
staged write.
+ assertEquals(expected, store.getPosition(), "getPosition before
commit (uncommitted)");
+
+ // after commit, committed position populated
+ store.commit(Map.of());
+ assertEquals(expected, store.getPosition(), "getPosition after
commit");
+ } finally {
+ store.close();
+ }
+ }
+
private InMemorySessionStore openTransactionalSessionStore() {
final Properties props = StreamsTestUtils.getStreamsConfig();
props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
index be6981ca9f5..72af0085de0 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
@@ -408,6 +408,46 @@ public class InMemoryWindowStoreTest extends
AbstractWindowBytesStoreTest {
}
}
+ @Test
+ public void shouldReportUncommittedPositionForTransactionalStore() {
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ props.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(props)
+ );
+ final InMemoryWindowStore store = new InMemoryWindowStore(
+ "txn-pos-window-store", RETENTION_PERIOD, WINDOW_SIZE, false,
"scope");
+ store.init(ctx, store);
+ try {
+ ctx.setRecordContext(new ProcessorRecordContext(0, 1, 0, "topic",
new RecordHeaders()));
+ store.put(Bytes.wrap("k".getBytes()), "v".getBytes(), 0L);
+
+ final Position expected = Position.fromMap(mkMap(mkEntry("topic",
mkMap(mkEntry(0, 1L)))));
+
+ // READ_UNCOMMITTED query should expose the staged position before
commit
+ final QueryResult<?> uncommitted = store.query(
+ WindowKeyQuery.withKeyAndWindowStartRange(
+ Bytes.wrap("k".getBytes()), Instant.ofEpochMilli(0),
Instant.ofEpochMilli(WINDOW_SIZE)),
+ PositionBound.unbounded(),
+ new QueryConfig(false));
+ assertEquals(expected, uncommitted.getPosition(),
"READ_UNCOMMITTED query position");
+
+ // getPosition() reports the uncommitted (committed + staged)
position, mirroring
+ // RocksDBStore, so the changelog consistency vector reflects the
staged write.
+ assertEquals(expected, store.getPosition(), "getPosition before
commit (uncommitted)");
+
+ // after commit, committed position populated
+ store.commit(java.util.Map.of());
+ assertEquals(expected, store.getPosition(), "getPosition after
commit");
+ } finally {
+ store.close();
+ }
+ }
+
private InMemoryWindowStore openTransactionalWindowStore() {
final Properties props = StreamsTestUtils.getStreamsConfig();
props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
