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 360b381cfb7 KAFKA-20760: Stage position updates in
InMemoryKeyValueStore transaction buffer (#22743)
360b381cfb7 is described below
commit 360b381cfb743d348e84641ebeeaffc8ca99b066
Author: Nick Telford <[email protected]>
AuthorDate: Fri Jul 3 01:48:25 2026 +0100
KAFKA-20760: Stage position updates in InMemoryKeyValueStore transaction
buffer (#22743)
`InMemoryKeyValueStore`, when transactional, did not correctly track its
`Position` across the transaction boundary, which broke
interactive-query position consistency in two ways.
**Staging.** `put`/`putAll` applied position updates directly to the
store's committed `Position` instead of holding them pending alongside
the staged writes, so the position advanced before the corresponding
data was committed. Position updates are now accumulated in a
`pendingPosition` on `InMemoryTransactionBuffer` during staging and
merged into the committed position only in `commitStagedWrites`.
`discardPendingBatch` (previously a no-op) resets `pendingPosition` so
aborted batches leave no trace. Under READ_UNCOMMITTED, `query` merges
the pending deltas so staged writes remain visible, mirroring
`RocksDBStore`.
**Changelog consistency vector.** `getPosition()` returned only the
committed position. Because `ChangeLoggingKeyValueBytesStore` stamps
each changelog record's consistency vector with `getPosition()` at put
time, a transactional store's changelog records carried a stale (empty)
vector, and the position could not be rebuilt when the store was
restored from its changelog on restart. `getPosition()` now returns the
uncommitted position (committed merged with the buffer's pending
deltas), matching the behaviour `RocksDBStore` already had.
Adds a unit test covering the uncommitted-position behaviour. End-to-end
restart coverage is added separately via the parameterized
`PositionRestartIntegrationTest`.
Part of a series of fixes for interactive-query position consistency
with transactional state stores.
Reviewers: Bill Bejeck <[email protected]>
---
.../state/internals/InMemoryKeyValueStore.java | 45 +++++++++++++++++-----
.../state/internals/InMemoryTransactionBuffer.java | 33 +++++++++++++++-
.../state/internals/InMemoryKeyValueStoreTest.java | 38 ++++++++++++++++++
3 files changed, 106 insertions(+), 10 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
index 0a1f94ef321..fcf570a101a 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
@@ -125,6 +125,16 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
@Override
public Position getPosition() {
+ // Mirror RocksDBStore#getPosition: report the uncommitted position
(committed + staged) so the
+ // changelog consistency vector, which ChangeLogging*BytesStore writes
at put() time via
+ // getPosition(), reflects the input position of the staged write.
Otherwise a transactional
+ // store's changelog records carry the stale committed position and
the position cannot be
+ // rebuilt when the store is restored from its changelog.
+ if (transactionBuffer != null) {
+ synchronized (position) {
+ return
position.copy().merge(transactionBuffer.pendingPosition());
+ }
+ }
return position;
}
@@ -133,14 +143,26 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
final PositionBound positionBound,
final QueryConfig config) {
- return StoreQueryUtils.handleBasicQueries(
- query,
- positionBound,
- config,
- this,
- position,
- context
- );
+ synchronized (position) {
+ // Mirror RocksDBStore#query: under READ_UNCOMMITTED, expose the
writes staged in the
+ // transaction buffer since the last commit by merging the
buffer's pending position
+ // deltas into a copy of the committed position. READ_COMMITTED
(and the
+ // non-transactional store) query the committed position directly.
+ final Position queryPosition;
+ if (transactionBuffer != null && config.getIsolationLevel() ==
IsolationLevel.READ_UNCOMMITTED) {
+ queryPosition =
position.copy().merge(transactionBuffer.pendingPosition());
+ } else {
+ queryPosition = position;
+ }
+ return StoreQueryUtils.handleBasicQueries(
+ query,
+ positionBound,
+ config,
+ this,
+ queryPosition,
+ context
+ );
+ }
}
@Override
@@ -167,6 +189,7 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
public synchronized void put(final Bytes key, final byte[] value) {
if (transactionBuffer != null) {
transactionBuffer.stage(key, value);
+ transactionBuffer.updatePosition(context);
return;
}
putInternal(key, value);
@@ -199,6 +222,7 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
if (transactionBuffer != null) {
for (final KeyValue<Bytes, byte[]> entry : entries) {
transactionBuffer.stage(entry.key, entry.value);
+ transactionBuffer.updatePosition(context);
}
return;
}
@@ -365,7 +389,10 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
void commitStagedWrites() {
if (transactionBuffer != null) {
- transactionBuffer.commit();
+ synchronized (position) {
+ transactionBuffer.mergePendingPositionInto(position);
+ transactionBuffer.commit();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
index 26a8ef8f61c..9de1f9c62a1 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
@@ -18,6 +18,8 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.query.Position;
import java.util.Map;
import java.util.NavigableMap;
@@ -42,6 +44,7 @@ import java.util.TreeMap;
class InMemoryTransactionBuffer extends AbstractTransactionBuffer<Bytes> {
private final NavigableMap<Bytes, byte[]> baseMap;
+ private Position pendingPosition = Position.emptyPosition();
InMemoryTransactionBuffer(final NavigableMap<Bytes, byte[]> baseMap) {
this.baseMap = baseMap;
@@ -127,7 +130,35 @@ class InMemoryTransactionBuffer extends
AbstractTransactionBuffer<Bytes> {
@Override
void discardPendingBatch() {
- // no-op — no backend batch to discard
+ pendingPosition = Position.emptyPosition();
+ }
+
+ void updatePosition(final StateStoreContext stateStoreContext) {
+ snapshotLock.writeLock().lock();
+ try {
+ StoreQueryUtils.updatePosition(pendingPosition, stateStoreContext);
+ } finally {
+ snapshotLock.writeLock().unlock();
+ }
+ }
+
+ Position pendingPosition() {
+ snapshotLock.readLock().lock();
+ try {
+ return pendingPosition.copy();
+ } finally {
+ snapshotLock.readLock().unlock();
+ }
+ }
+
+ void mergePendingPositionInto(final Position committed) {
+ snapshotLock.writeLock().lock();
+ try {
+ committed.merge(pendingPosition);
+ pendingPosition = Position.emptyPosition();
+ } finally {
+ snapshotLock.writeLock().unlock();
+ }
}
/**
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
index 894d88674fc..93a84a748a4 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
@@ -600,6 +600,44 @@ public class InMemoryKeyValueStoreTest extends
AbstractKeyValueStoreTest {
}
}
+ @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 InMemoryKeyValueStore store = new
InMemoryKeyValueStore("txn-pos-store");
+ store.init(ctx, store);
+ try {
+ ctx.setRecordContext(new ProcessorRecordContext(0, 1, 0, "topic",
new RecordHeaders()));
+ store.put(bytesKey("key1"), bytesValue("value1"));
+
+ final Position expected = Position.fromMap(mkMap(mkEntry("topic",
mkMap(mkEntry(0, 1L)))));
+
+ // READ_UNCOMMITTED query should expose the staged position before
commit
+ final org.apache.kafka.streams.query.QueryResult<?> uncommitted =
store.query(
+ org.apache.kafka.streams.query.RangeQuery.withNoBounds(),
+ org.apache.kafka.streams.query.PositionBound.unbounded(),
+ new org.apache.kafka.streams.query.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 InMemoryKeyValueStore openTransactionalStore() {
final Properties props = StreamsTestUtils.getStreamsConfig();
props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
