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 b30cd5f6a9d KAFKA-20760: Break out PositionRestartIntegrationTest into
sub-class for transactional stores (#22823)
b30cd5f6a9d is described below
commit b30cd5f6a9dbda39f0dc59643f215326ad81d8ad
Author: Bill Bejeck <[email protected]>
AuthorDate: Thu Jul 16 14:49:51 2026 -0400
KAFKA-20760: Break out PositionRestartIntegrationTest into sub-class for
transactional stores (#22823)
The transactional (KIP-892) Position-across-restart coverage previously
added to PositionRestartIntegrationTest expanded the parameter matrix
across the full cache x log x store x kind cross-product, roughly
doubling the case count and runtime and making the test prone to
timeouts.
This splits the concern:
- PositionRestartIntegrationTest returns to its original
non-transactional matrix,
with small hooks (an overridable `data()` source and a
`transactional()` flag)
so the machinery can be reused
- TransactionalPositionRestartIntegrationTest extends it and supplies a
compact
covering set for the transactional path: every persistent and
in-memory store
family (transactional state stores are supported for both), log
enabled
throughout, rotating cache and kind so both values are exercised
without a full
cross-product, all under exactly-once-v2.
Reviewers: Nick Telford <[email protected]>, Matthias J. Sax
<[email protected]>
---
.../PositionRestartIntegrationTest.java | 48 ++++++++------
...ransactionalPositionRestartIntegrationTest.java | 74 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 19 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 612f9633256..66d65f27423 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
@@ -69,6 +69,7 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -101,6 +102,9 @@ import static org.hamcrest.Matchers.is;
@Tag("integration")
@Timeout(600)
+// PER_CLASS lets the parameter source `data()` be a non-static, overridable
method so subclasses
+// (e.g. the transactional variant) can supply a different matrix while
reusing all the machinery below.
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class PositionRestartIntegrationTest {
private static final Logger LOG =
LoggerFactory.getLogger(PositionRestartIntegrationTest.class);
private static final long SEED = new Random().nextLong();
@@ -252,19 +256,22 @@ public class PositionRestartIntegrationTest {
}
}
- public static Stream<Arguments> data() {
+ /** Whether the store-under-test is configured transactional (KIP-892).
Overridden by the transactional variant. */
+ protected boolean transactional() {
+ return false;
+ }
+
+ protected Stream<Arguments> data() {
LOG.info("Generating test cases according to random seed: {}", SEED);
final List<Arguments> values = new ArrayList<>();
for (final boolean cacheEnabled : Arrays.asList(true, false)) {
for (final boolean logEnabled : Arrays.asList(true, false)) {
- 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));
- }
+ 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));
}
}
}
@@ -365,8 +372,8 @@ public class PositionRestartIntegrationTest {
@ParameterizedTest
@MethodSource("data")
- 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);
+ public void verifyStore(final boolean cache, final boolean log, final
StoresToTest storeToTest, final String kind) {
+ final Properties streamsConfig = streamsConfiguration(cache, log,
storeToTest.name(), kind);
final StreamsBuilder streamsBuilder = getStreamBuilder(cache, log,
storeToTest, kind);
kafkaStreams = IntegrationTestUtils.getStartedStreams(streamsConfig,
streamsBuilder, true);
@@ -650,14 +657,13 @@ public class PositionRestartIntegrationTest {
.process(processorSupplier, sessionStoreStoreBuilder.name());
}
- private static Properties streamsConfiguration(final boolean cache,
- final boolean log,
- final boolean transactional,
- final String supplier,
- final String kind) {
+ protected Properties streamsConfiguration(final boolean cache,
+ final boolean log,
+ final String supplier,
+ final String kind) {
final String safeTestName =
- PositionRestartIntegrationTest.class.getName() + "-" + cache + "-"
+ log + "-" + transactional + "-"
- + supplier + "-" + kind;
+ getClass().getName() + "-" + cache + "-" + log + "-"
+ + supplier + "-" + kind + "-" + transactional();
final Properties config = new Properties();
config.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG,
StreamsConfig.OPTIMIZE);
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName);
@@ -671,7 +677,11 @@ 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);
+ if (transactional()) {
+ // Transactional state stores (KIP-892) are only meaningful under
exactly-once.
+ config.put(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG, true);
+ config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ }
return config;
}
}
\ No newline at end of file
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TransactionalPositionRestartIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TransactionalPositionRestartIntegrationTest.java
new file mode 100644
index 00000000000..716eccefb0b
--- /dev/null
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TransactionalPositionRestartIntegrationTest.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.provider.Arguments;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+/**
+ * Exercises the transactional (KIP-892) Position-across-restart path, reusing
all of
+ * {@link PositionRestartIntegrationTest}'s machinery (topology building,
produce, restart, IQ position
+ * verification) by overriding two hooks: {@link #transactional()} (turns on
+ * {@code enable.transactional.statestores} + {@code exactly_once_v2}) and
{@link #data()} (supplies a
+ * transactional-only matrix).
+ *
+ * <p>Keeping this separate from the parent keeps the (large)
non-transactional matrix and the transactional
+ * matrix in independent test targets, each with its own time budget, rather
than multiplying one big
+ * cross-product.
+ */
+@Tag("integration")
+@Timeout(600)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class TransactionalPositionRestartIntegrationTest extends
PositionRestartIntegrationTest {
+
+ @Override
+ protected boolean transactional() {
+ return true;
+ }
+
+ @Override
+ protected Stream<Arguments> data() {
+ // Every persistent and in-memory store family, crossed with caching
on/off. log=true throughout so
+ // in-memory stores survive the restart via changelog restore; kind is
rotated (not crossed) since it
+ // is store-level-independent for the Position API.
+ final List<StoresToTest> stores = List.of(
+ StoresToTest.ROCKS_KV,
+ StoresToTest.TIME_ROCKS_KV,
+ StoresToTest.ROCKS_WINDOW,
+ StoresToTest.TIME_ROCKS_WINDOW,
+ StoresToTest.ROCKS_SESSION,
+ StoresToTest.IN_MEMORY_KV,
+ StoresToTest.IN_MEMORY_WINDOW,
+ StoresToTest.IN_MEMORY_SESSION
+ );
+ final List<Arguments> values = new ArrayList<>();
+ int i = 0;
+ for (final StoresToTest store : stores) {
+ for (final boolean cache : List.of(false, true)) {
+ final String kind = (i++ % 2 == 0) ? "DSL" : "PAPI";
+ values.add(Arguments.of(cache, true, store, kind));
+ }
+ }
+ return values.stream();
+ }
+}
