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 59596de362b KAFKA-20760: Stage position updates in
InMemorySessionStore transaction buffer (#22745)
59596de362b is described below
commit 59596de362ba4d3b6e492681e2c8381393c63328
Author: Nick Telford <[email protected]>
AuthorDate: Fri Jul 3 01:49:46 2026 +0100
KAFKA-20760: Stage position updates in InMemorySessionStore transaction
buffer (#22745)
`InMemorySessionStore`, when transactional, did not correctly track its
`Position` across the transaction boundary, breaking interactive-query
position consistency in two ways. The root cause is the same as the
parallel fix for `InMemoryKeyValueStore`.
**Staging.** `put` called `StoreQueryUtils.updatePosition` directly on
the store's committed `Position` on every write, bypassing the
transaction boundary, so IQv2 position-bound queries could observe
position advances corresponding to records not yet visible. Position
updates are now staged into a `pendingPosition` field on
`InMemorySessionTransactionBuffer` under the existing `snapshotLock`,
and merged into the committed position at commit time via
`mergePendingPositionInto`. `discardPendingBatch` resets the pending
position on abort. 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 `ChangeLoggingSessionBytesStore` stamps each
changelog record's consistency vector with `getPosition()` at put time,
a transactional store's changelog records carried a stale (empty)
vector, so the position could not be rebuilt when the store was restored
from its changelog. `getPosition()` now returns the uncommitted
position, matching the behaviour `RocksDBStore` already had.
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]>
Co-authored-by: Claude Sonnet 4.6 (1M context) <[email protected]>
---
.../state/internals/InMemorySessionStore.java | 49 +++++++++++++++++-----
.../InMemorySessionTransactionBuffer.java | 33 ++++++++++++++-
2 files changed, 71 insertions(+), 11 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
index 0b36f5b11c4..fa81a903dfa 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
@@ -181,6 +181,16 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
@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;
}
@@ -205,7 +215,11 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
putInternal(sessionKey, aggregate);
}
- StoreQueryUtils.updatePosition(position, stateStoreContext);
+ if (transactionBuffer != null) {
+ transactionBuffer.updatePosition(stateStoreContext);
+ } else {
+ StoreQueryUtils.updatePosition(position, stateStoreContext);
+ }
}
}
@@ -462,14 +476,26 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
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
@@ -563,7 +589,10 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
@Override
public void commit(final Map<TopicPartition, Long> changelogOffsets) {
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/InMemorySessionTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
index 00d4e8246fd..cfaf79a274b 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
@@ -19,6 +19,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.kstream.Windowed;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.state.KeyValueIterator;
import java.util.Map;
@@ -34,6 +36,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
class InMemorySessionTransactionBuffer extends
AbstractTransactionBuffer<InMemorySessionTransactionBuffer.SessionEntryKey> {
private final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeMap;
+ private Position pendingPosition = Position.emptyPosition();
InMemorySessionTransactionBuffer(
final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeMap) {
@@ -226,6 +229,34 @@ class InMemorySessionTransactionBuffer extends
AbstractTransactionBuffer<InMemor
forward));
}
+ 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();
+ }
+ }
+
@Override
void flushToBase() {
for (final Map.Entry<SessionEntryKey, Optional<byte[]>> entry :
pendingWrites.entrySet()) {
@@ -256,7 +287,7 @@ class InMemorySessionTransactionBuffer extends
AbstractTransactionBuffer<InMemor
@Override
void discardPendingBatch() {
- // no-op — no backend batch to discard
+ pendingPosition = Position.emptyPosition();
}
/**
