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 2a0efd08b98 KAFKA-20749: Stop offset writes leaking into data
iteration (#22710)
2a0efd08b98 is described below
commit 2a0efd08b984738cc054e916e193521d5a57ede2
Author: Nick Telford <[email protected]>
AuthorDate: Tue Jun 30 20:06:05 2026 +0100
KAFKA-20749: Stop offset writes leaking into data iteration (#22710)
A KStream-KStream OUTER join crashes with a `SerializationException`
under `exactly_once_v2` when `enable.transactional.statestores=true`. As
soon as the join emits a non-joined (outer) record, the StreamThread
dies and the outer result is never produced.
`RocksDBTransactionBuffer` stages every write into a single in-memory
`pendingWrites` map regardless of which column family the write targets.
Writes to the offsets column family (the persisted `Position`, written
under the `"position"` key) therefore landed in the same map as data-CF
writes, so subsequent data-CF `range()`/`all()`/`prefixScan()` scans
surfaced the offsets entry and tried to deserialize the serialized
`Position` as a data record. The outer-join `ListValueStore` could not
deserialize it as a value list and threw. The `get()` path was already
gated against the offsets CF; the iterator paths were not.
Non-transactional stores write straight to the offsets CF and never leak
through the buffer, which is why this is transactional-only.
The fix stages a write into `pendingWrites` only when its column family
is the data column family (`cfHandle`). Offset-CF writes still go to the
underlying `writeBatch` so they commit atomically with the data, but
they no longer pollute data-CF iteration.
A `RocksDBStoreTest` regression test asserts that
`all`/`range`/`prefixScan` return only the data key after
`writePosition()`, and a new integration test reproduces the original
outer-join crash end to end (parameterized over
`enable.transactional.statestores`, so the non-transactional case acts
as a passing control).
The tests in this PR originated from @bbejeck's #22709, which fixes the
same bug via a separate `stageOffsetWrite` path; this PR takes a smaller
approach that guards the single `stage` entry point and needs no changes
to `RocksDBStore`.
Reviewers: Bill Bejeck <[email protected]>
Co-authored-by: Bill Bejeck <[email protected]>
---
...OuterJoinTransactionalStoreIntegrationTest.java | 222 +++++++++++++++++++++
.../state/internals/RocksDBTransactionBuffer.java | 6 +-
.../streams/state/internals/RocksDBStoreTest.java | 34 ++++
.../internals/RocksDBTransactionBufferTest.java | 9 +-
4 files changed, 264 insertions(+), 7 deletions(-)
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamKStreamOuterJoinTransactionalStoreIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamKStreamOuterJoinTransactionalStoreIntegrationTest.java
new file mode 100644
index 00000000000..ef3f6302e7a
--- /dev/null
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamKStreamOuterJoinTransactionalStoreIntegrationTest.java
@@ -0,0 +1,222 @@
+/*
+ * 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.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.JoinWindows;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.StreamJoined;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.time.Duration;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived;
+import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Reproduces a bug KAFKA-20749: a KStream-KStream OUTER join crashes when it
+ * emits non-joined (outer) records under {@code exactly_once_v2} <em>and</em>
+ * {@code enable.transactional.statestores=true}. Reading the outer-join buffer
+ * ({@link org.apache.kafka.streams.state.internals.ListValueStore}) throws a
{@code SerializationException}
+ * because the transactional persistent-store value round-trip does not
preserve the exact bytes that
+ * {@code ListValueStore} wrote with {@code Serdes.ListSerde(ArrayList.class,
Serdes.ByteArray())}.
+ *
+ * <p>The test is parameterized over {@code enable.transactional.statestores}:
the {@code false} case is a
+ * control that passes today (same topology, non-transactional stores), while
the {@code true} case fails on
+ * current trunk (the {@code SerializationException} kills the StreamThread,
so the outer result is never
+ * emitted) and passes once the bug is fixed.
+ */
+@Timeout(600)
+@Tag("integration")
+public class KStreamKStreamOuterJoinTransactionalStoreIntegrationTest {
+
+ private static final int NUM_BROKERS = 1;
+ public static final EmbeddedKafkaCluster CLUSTER = new
EmbeddedKafkaCluster(NUM_BROKERS);
+
+ private String applicationId;
+ private String leftTopic;
+ private String rightTopic;
+ private String outputTopic;
+ private Properties streamsConfig;
+ private KafkaStreams streams;
+
+ @BeforeAll
+ public static void startCluster() throws Exception {
+ CLUSTER.start();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ CLUSTER.stop();
+ }
+
+ @BeforeEach
+ public void before(final TestInfo testInfo) throws Exception {
+ applicationId = "outer-join-txn-store-" + safeUniqueTestName(testInfo);
+ leftTopic = applicationId + "-left";
+ rightTopic = applicationId + "-right";
+ outputTopic = applicationId + "-output";
+ CLUSTER.createTopic(leftTopic, 1, 1);
+ CLUSTER.createTopic(rightTopic, 1, 1);
+ CLUSTER.createTopic(outputTopic, 1, 1);
+
+ streamsConfig = new Properties();
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
+ streamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
+ streamsConfig.put(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath());
+ streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.String().getClass());
+ streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.String().getClass());
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ // Disable the store cache so emitted records are not
buffered/absorbed by the cache.
+ streamsConfig.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
+ // Emit non-joined outer records on every stream-time advance (no
wall-clock throttle), so a single
+ // window-advancing record deterministically triggers
emitNonJoinedOuterRecords over the buffer.
+
streamsConfig.put(StreamsConfig.InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX,
0L);
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ if (streams != null) {
+ streams.close(Duration.ofSeconds(30));
+ streams.cleanUp();
+ }
+ CLUSTER.deleteAllTopics();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldEmitOuterRecordWithTransactionalStateStores(final
boolean transactionalStateStores) throws Exception {
+ streamsConfig.put(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
transactionalStateStores);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ final KStream<String, String> leftStream = builder.stream(leftTopic);
+ final KStream<String, String> rightStream = builder.stream(rightTopic);
+
+ // Default (persistent RocksDB) stores; with
enable.transactional.statestores=true the outer-join
+ // ListValueStore is wrapped by the transactional accessor — the path
that corrupts the value round-trip.
+ leftStream.outerJoin(
+ rightStream,
+ (leftValue, rightValue) -> leftValue + "/" + rightValue,
+ JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofSeconds(60)),
+ StreamJoined.with(Serdes.String(), Serdes.String(),
Serdes.String())
+ ).to(outputTopic);
+
+ streams = new KafkaStreams(builder.build(), streamsConfig);
+
+ final AtomicReference<Throwable> uncaught = new AtomicReference<>();
+ streams.setUncaughtExceptionHandler(throwable -> {
+ uncaught.compareAndSet(null, throwable);
+ return
StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
+ });
+
+ startApplicationAndWaitUntilRunning(streams);
+
+ // Unmatched left record -> goes into the outer-join buffer
(ListValueStore), no matching right.
+ produceRecord(leftTopic, "A", "L1", 1000L);
+
+ // Advance stream time well past the 60s window so
emitNonJoinedOuterRecords() runs and reads the
+ // buffer. This is where the SerializationException is thrown when
stores are transactional.
+ produceRecord(leftTopic, "B", "L2", 62000L);
+
+ final List<KeyValue<String, String>> results;
+ try {
+ results = waitUntilMinKeyValueRecordsReceived(getConsumerConfig(),
outputTopic, 1, 60000);
+ } catch (final AssertionError timeout) {
+ // Most informative failure: surface the StreamThread crash that
prevented the emit.
+ final Throwable crash = uncaught.get();
+ if (crash != null) {
+ throw new AssertionError(
+ "emitNonJoinedOuterRecords failed to emit the outer
record; StreamThread crashed with:\n"
+ + stackTraceToString(crash), crash);
+ }
+ throw timeout;
+ }
+
+ assertNull(uncaught.get(),
+ uncaught.get() == null ? "" : "StreamThread crashed:\n" +
stackTraceToString(uncaught.get()));
+
+ final KeyValue<String, String> outerResult = results.stream()
+ .filter(kv -> "A".equals(kv.key))
+ .findFirst()
+ .orElse(null);
+
+ assertEquals(KeyValue.pair("A", "L1/null"), outerResult,
+ "The unmatched left record A should be emitted as the outer result
(A, \"L1/null\")");
+ }
+
+ private void produceRecord(final String topic, final String key, final
String value, final long timestamp) {
+ final Properties producerConfig = new Properties();
+ producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
+ producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
+
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ topic,
+ List.of(new KeyValue<>(key, value)),
+ producerConfig,
+ timestamp
+ );
+ }
+
+ private Properties getConsumerConfig() {
+ final Properties consumerConfig = new Properties();
+ consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
+ consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-" +
applicationId);
+ consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
+ consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
+ consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest");
+ return consumerConfig;
+ }
+
+ private static String stackTraceToString(final Throwable throwable) {
+ final StringWriter sw = new StringWriter();
+ throwable.printStackTrace(new PrintWriter(sw));
+ return sw.toString();
+ }
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java
index 088d8d412b8..29da08db2d5 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java
@@ -95,8 +95,10 @@ class RocksDBTransactionBuffer extends
AbstractTransactionBuffer<Bytes> {
void stage(final ColumnFamilyHandle cf, final Bytes key, final byte[]
value) {
snapshotLock.writeLock().lock();
try {
- pendingWrites.put(key, Optional.ofNullable(value));
- pendingWritesBytes += estimateKeySize(key) + (value != null ?
value.length : 0);
+ if (cf == cfHandle) {
+ pendingWrites.put(key, Optional.ofNullable(value));
+ pendingWritesBytes += estimateKeySize(key) + (value != null ?
value.length : 0);
+ }
try {
if (value != null) {
writeBatch.put(cf, key.get(), value);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
index 70edbf9345c..fd001efa4ee 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
@@ -1616,6 +1616,40 @@ public class RocksDBStoreTest extends
AbstractKeyValueStoreTest {
assertTrue(rocksDBStore.getPosition().getTopics().isEmpty());
}
+ @Test
+ public void offsetColumnFamilyWritesShouldNotLeakIntoDataIteration() {
+ // Regression test for the transactional-store outer-join bug -
KAFKA-20749
+ rocksDBStore.close();
+ final InternalMockProcessorContext<?, ?> eosContext =
getTransactionalEOSProcessorContext(dir);
+ rocksDBStore = getRocksDBStore();
+ rocksDBStore.init(eosContext, rocksDBStore);
+
+ eosContext.setRecordContext(new ProcessorRecordContext(0, 1L, 0,
"input", new RecordHeaders()));
+ rocksDBStore.put(new Bytes(stringSerializer.serialize(null, "k1")),
stringSerializer.serialize(null, "v1"));
+
+ // Persist the Position into the offsets column family; this stays
staged in the buffer (not committed).
+ rocksDBStore.writePosition();
+
+ // Data-CF scans must surface only the data key, never the offsets-CF
"position" entry.
+ try (KeyValueIterator<Bytes, byte[]> it = rocksDBStore.all()) {
+ assertEquals(List.of("k1"), keysOf(it));
+ }
+ try (KeyValueIterator<Bytes, byte[]> it = rocksDBStore.range(null,
null)) {
+ assertEquals(List.of("k1"), keysOf(it));
+ }
+ try (KeyValueIterator<Bytes, byte[]> it = rocksDBStore.prefixScan("k",
stringSerializer)) {
+ assertEquals(List.of("k1"), keysOf(it));
+ }
+ }
+
+ private List<String> keysOf(final KeyValueIterator<Bytes, byte[]> it) {
+ final List<String> keys = new ArrayList<>();
+ while (it.hasNext()) {
+ keys.add(stringDeserializer.deserialize(null,
it.next().key.get()));
+ }
+ return keys;
+ }
+
public static class TestingBloomFilterRocksDBConfigSetter implements
RocksDBConfigSetter {
static boolean bloomFiltersSet;
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBufferTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBufferTest.java
index b0f7f11e37b..db15e266c7a 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBufferTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBufferTest.java
@@ -420,9 +420,8 @@ public class RocksDBTransactionBufferTest {
public void shouldStageWriteToOtherCF() throws RocksDBException {
buffer.stage(otherCfHandle, key("x"), val("other-x"));
- // Shared read buffer sees the staged value
- assertTrue(buffer.get(key("x")).isPresent());
- assertArrayEquals(val("other-x"), buffer.get(key("x")).get());
+ // Non-primary CF writes are not visible in the staging read buffer
+ assertNull(buffer.get(key("x")));
// Not yet flushed to RocksDB
assertNull(db.get(otherCfHandle, key("x").get()));
@@ -437,8 +436,8 @@ public class RocksDBTransactionBufferTest {
buffer.stage(otherCfHandle, key("x"), null);
- // Shared read buffer shows tombstone
- assertEquals(Optional.empty(), buffer.get(key("x")));
+ // Non-primary CF deletes are not visible in the staging read buffer
+ assertNull(buffer.get(key("x")));
// Not yet flushed
assertArrayEquals(val("other-x"), db.get(otherCfHandle,
key("x").get()));
