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 569fdcb53e3 KAFKA-20760: Stage position updates in InMemoryWindowStore
transaction buffer (#22744)
569fdcb53e3 is described below
commit 569fdcb53e307f6c31e45efbe54d12899dcef078
Author: Nick Telford <[email protected]>
AuthorDate: Fri Jul 3 01:48:56 2026 +0100
KAFKA-20760: Stage position updates in InMemoryWindowStore transaction
buffer (#22744)
`InMemoryWindowStore`, when transactional, did not correctly track its
`Position` across the transaction boundary, breaking interactive-query
position consistency in two ways.
**Staging.** `put` called `StoreQueryUtils.updatePosition` directly on
the store's committed `Position`, making uncommitted records queryable
by position and violating the isolation guarantee. Position updates are
now accumulated in a `pendingPosition` field on
`InMemoryWindowTransactionBuffer` under the existing `snapshotLock`, and
`mergePendingPositionInto` merges them into the committed position at
commit time. `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 `ChangeLoggingWindowBytesStore` 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]>
---
.../state/internals/InMemoryWindowStore.java | 49 +++++++++++++++++-----
.../internals/InMemoryWindowTransactionBuffer.java | 33 ++++++++++++++-
2 files changed, 71 insertions(+), 11 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
index 2a6ecbdb694..aa61bbc0a48 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
@@ -175,6 +175,16 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
@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;
}
@@ -200,7 +210,11 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
putInternal(key, value, windowStartTimestamp);
}
- StoreQueryUtils.updatePosition(position, internalProcessorContext);
+ if (transactionBuffer != null) {
+ transactionBuffer.updatePosition(internalProcessorContext);
+ } else {
+ StoreQueryUtils.updatePosition(position,
internalProcessorContext);
+ }
}
}
@@ -474,14 +488,26 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final PositionBound positionBound,
final QueryConfig config) {
- return StoreQueryUtils.handleBasicQueries(
- query,
- positionBound,
- config,
- this,
- position,
- internalProcessorContext
- );
+ 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,
+ internalProcessorContext
+ );
+ }
}
@Override
@@ -564,7 +590,10 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
@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/InMemoryWindowTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
index 03fe8ca35b7..494ea732da6 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
@@ -17,6 +17,8 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.query.Position;
import java.util.Map;
import java.util.Objects;
@@ -32,6 +34,7 @@ class InMemoryWindowTransactionBuffer extends
AbstractTransactionBuffer<InMemory
private final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> segmentMap;
private final boolean retainDuplicates;
+ private Position pendingPosition = Position.emptyPosition();
InMemoryWindowTransactionBuffer(
final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> segmentMap,
@@ -204,6 +207,34 @@ class InMemoryWindowTransactionBuffer extends
AbstractTransactionBuffer<InMemory
keyFrom, keyTo, timeRange.entrySet().iterator(), retainDuplicates,
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<WindowEntryKey, Optional<byte[]>> entry :
pendingWrites.entrySet()) {
@@ -225,7 +256,7 @@ class InMemoryWindowTransactionBuffer extends
AbstractTransactionBuffer<InMemory
@Override
void discardPendingBatch() {
- // no-op — no backend batch to discard
+ pendingPosition = Position.emptyPosition();
}
}
