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 d2328ff720b KAFKA-20495: Add transactional support to in-memory window
and session stores (#22652)
d2328ff720b is described below
commit d2328ff720ba4337d50bbdea01342fb4141dcd3f
Author: Nick Telford <[email protected]>
AuthorDate: Thu Jun 25 22:06:51 2026 +0100
KAFKA-20495: Add transactional support to in-memory window and session
stores (#22652)
In-memory window and session stores need the same transactional,
isolation-aware behaviour KIP-892 introduced for the key-value store:
when transactional state stores are enabled (EOS), writes must be staged
in a transaction buffer and only applied to the store on commit, and
interactive queries must be able to read at `READ_COMMITTED` (ignoring
staged writes) or `READ_UNCOMMITTED`.
This adds `InMemoryWindowTransactionBuffer` and
`InMemorySessionTransactionBuffer`, each extending
`AbstractTransactionBuffer` with a composite key that encodes the
store's ordering — timestamp/key for windows, endTime/key/startTime for
sessions. `InMemoryWindowStore` and `InMemorySessionStore` route reads
and writes through their buffer when transactional state stores are
enabled, and report their staged size via
`approximateNumUncommittedBytes()`.
Each buffer implements `newBaseSnapshotIterator` so non-owner
(interactive-query) reads get a point-in-time view: the bounded range of
the backing segment map is eagerly deep-copied under the snapshot read
lock, isolating the returned iterator from concurrent owner mutation.
Reviewers: Bill Bejeck <[email protected]>
---
.../state/internals/InMemorySessionStore.java | 262 ++++++++++++-
.../InMemorySessionTransactionBuffer.java | 278 ++++++++++++++
.../state/internals/InMemoryWindowStore.java | 405 ++++++++++++++++++++-
.../internals/InMemoryWindowTransactionBuffer.java | 203 +++++++++++
.../internals/AbstractSessionBytesStoreTest.java | 12 +-
.../internals/AbstractWindowBytesStoreTest.java | 14 +-
.../InMemoryTransactionalSessionStoreTest.java | 77 ++++
.../InMemoryTransactionalWindowStoreTest.java | 125 +++++++
8 files changed, 1342 insertions(+), 34 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 dd3f083aa10..b44389772bc 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
@@ -47,10 +47,12 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.function.Consumer;
import static
org.apache.kafka.streams.StreamsConfig.InternalConfig.IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED;
@@ -74,11 +76,13 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
private final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeMap = new
ConcurrentSkipListMap<>();
private final Set<InMemorySessionStoreIterator> openIterators =
ConcurrentHashMap.newKeySet();
+ private final Set<TransactionalSessionIterator> openTransactionalIterators
= ConcurrentHashMap.newKeySet();
private volatile boolean open = false;
private StateStoreContext stateStoreContext;
private final Position position;
+ private InMemorySessionTransactionBuffer transactionBuffer;
public InMemorySessionStore(
final String name,
@@ -133,19 +137,43 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
root,
(RecordBatchingStateRestoreCallback) records -> {
synchronized (position) {
+ long expiredRecords = 0;
for (final ConsumerRecord<byte[], byte[]> record :
records) {
-
put(SessionKeySchema.from(Bytes.wrap(record.key())), record.value());
+ final Windowed<Bytes> sessionKey =
SessionKeySchema.from(Bytes.wrap(record.key()));
+ final long windowEndTimestamp =
sessionKey.window().end();
+ observedStreamTime = Math.max(observedStreamTime,
windowEndTimestamp);
+ if (windowEndTimestamp <= observedStreamTime -
retentionPeriod) {
+ expiredRecords++;
+ } else {
+ // Write directly to the committed map:
restored records are already committed.
+ putInternal(sessionKey, record.value());
+ }
ChangelogRecordDeserializationHelper.applyChecksAndUpdatePosition(
record,
consistencyEnabled,
position
);
}
+ removeExpiredSegments();
+ if (expiredRecords > 0) {
+ if (expiredRecordSensor != null && context !=
null) {
+ expiredRecordSensor.record(expiredRecords,
context.currentSystemTimeMs());
+ }
+ LOG.warn("Skipping {} records for expired
segments.", expiredRecords);
+ }
}
}
);
}
open = true;
+
+ final boolean transactional = StreamsConfig.InternalConfig.getBoolean(
+ stateStoreContext.appConfigs(),
+ StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
+ false);
+ if (transactional) {
+ this.transactionBuffer = new
InMemorySessionTransactionBuffer(endTimeMap);
+ }
}
@Override
@@ -168,23 +196,38 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
expiredRecordSensor.record(1.0d,
context.currentSystemTimeMs());
}
LOG.warn("Skipping record for expired segment.");
+ } else if (transactionBuffer != null) {
+ transactionBuffer.stage(sessionKey, aggregate);
} else {
- if (aggregate != null) {
- endTimeMap.computeIfAbsent(windowEndTimestamp, t -> new
ConcurrentSkipListMap<>());
- final ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>> keyMap =
endTimeMap.get(windowEndTimestamp);
- keyMap.computeIfAbsent(sessionKey.key(), t -> new
ConcurrentSkipListMap<>());
-
keyMap.get(sessionKey.key()).put(sessionKey.window().start(), aggregate);
- } else {
- remove(sessionKey);
- }
+ putInternal(sessionKey, aggregate);
}
StoreQueryUtils.updatePosition(position, stateStoreContext);
}
}
+ private void putInternal(final Windowed<Bytes> sessionKey, final byte[]
aggregate) {
+ if (aggregate != null) {
+ final long windowEndTimestamp = sessionKey.window().end();
+ endTimeMap.computeIfAbsent(windowEndTimestamp, t -> new
ConcurrentSkipListMap<>());
+ final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long,
byte[]>> keyMap = endTimeMap.get(windowEndTimestamp);
+ keyMap.computeIfAbsent(sessionKey.key(), t -> new
ConcurrentSkipListMap<>());
+ keyMap.get(sessionKey.key()).put(sessionKey.window().start(),
aggregate);
+ } else {
+ removeFromBase(sessionKey);
+ }
+ }
+
@Override
public void remove(final Windowed<Bytes> sessionKey) {
+ if (transactionBuffer != null) {
+ transactionBuffer.stage(sessionKey, null);
+ return;
+ }
+ removeFromBase(sessionKey);
+ }
+
+ private void removeFromBase(final Windowed<Bytes> sessionKey) {
final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long,
byte[]>> keyMap = endTimeMap.get(sessionKey.window().end());
if (keyMap == null) {
return;
@@ -215,6 +258,13 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
// Only need to search if the record hasn't expired yet
if (sessionEndTime > observedStreamTime - retentionPeriod) {
+ if (transactionBuffer != null) {
+ final Optional<byte[]> staged = transactionBuffer.get(key,
sessionStartTime, sessionEndTime);
+ if (staged != null) {
+ return staged.orElse(null);
+ }
+ }
+
final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long,
byte[]>> keyMap = endTimeMap.get(sessionEndTime);
if (keyMap != null) {
final ConcurrentNavigableMap<Long, byte[]> startTimeMap =
keyMap.get(key);
@@ -231,6 +281,12 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
final long
latestSessionEndTime) {
removeExpiredSegments();
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(
+ null, null, Long.MAX_VALUE, earliestSessionEndTime,
latestSessionEndTime, true
+ );
+ }
+
final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimSubMap
= endTimeMap.subMap(earliestSessionEndTime, true,
latestSessionEndTime, true);
@@ -245,6 +301,12 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
removeExpiredSegments();
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(
+ key, key, latestSessionStartTime, earliestSessionEndTime,
Long.MAX_VALUE, true
+ );
+ }
+
return registerNewIterator(key,
key,
latestSessionStartTime,
@@ -260,6 +322,12 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
removeExpiredSegments();
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(
+ key, key, latestSessionStartTime, earliestSessionEndTime,
Long.MAX_VALUE, false
+ );
+ }
+
return registerNewIterator(
key,
key,
@@ -281,6 +349,12 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
return KeyValueIterators.emptyIterator();
}
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(
+ keyFrom, keyTo, latestSessionStartTime,
earliestSessionEndTime, Long.MAX_VALUE, true
+ );
+ }
+
return registerNewIterator(keyFrom,
keyTo,
latestSessionStartTime,
@@ -300,6 +374,12 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
return KeyValueIterators.emptyIterator();
}
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(
+ keyFrom, keyTo, latestSessionStartTime,
earliestSessionEndTime, Long.MAX_VALUE, false
+ );
+ }
+
return registerNewIterator(
keyFrom,
keyTo,
@@ -316,6 +396,10 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
removeExpiredSegments();
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(key, key, Long.MAX_VALUE,
0, Long.MAX_VALUE, true);
+ }
+
return registerNewIterator(key, key, Long.MAX_VALUE,
endTimeMap.entrySet().iterator(), true);
}
@@ -326,6 +410,10 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
removeExpiredSegments();
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(key, key, Long.MAX_VALUE,
0, Long.MAX_VALUE, false);
+ }
+
return registerNewIterator(key, key, Long.MAX_VALUE,
endTimeMap.descendingMap().entrySet().iterator(), false);
}
@@ -333,6 +421,10 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
keyFrom, final Bytes keyTo) {
removeExpiredSegments();
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(keyFrom, keyTo,
Long.MAX_VALUE, 0, Long.MAX_VALUE, true);
+ }
+
return registerNewIterator(keyFrom, keyTo, Long.MAX_VALUE,
endTimeMap.entrySet().iterator(), true);
}
@@ -340,6 +432,10 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final Bytes
keyFrom, final Bytes keyTo) {
removeExpiredSegments();
+ if (transactionBuffer != null) {
+ return newTransactionalSessionIterator(keyFrom, keyTo,
Long.MAX_VALUE, 0, Long.MAX_VALUE, false);
+ }
+
return registerNewIterator(
keyFrom, keyTo, Long.MAX_VALUE,
endTimeMap.descendingMap().entrySet().iterator(), false);
}
@@ -369,22 +465,41 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ if (transactionBuffer != null) {
+ return transactionBuffer.approximateNumUncommittedBytes();
+ }
+ return 0;
+ }
+
@Override
public void commit(final Map<TopicPartition, Long> changelogOffsets) {
- // do-nothing since it is in-memory
+ if (transactionBuffer != null) {
+ transactionBuffer.commit();
+ }
}
@Override
public void close() {
- if (openIterators.size() != 0) {
- LOG.warn("Closing {} open iterators for store {}",
openIterators.size(), name);
+ if (transactionBuffer != null) {
+ transactionBuffer.rollback();
+ }
+
+ final int openCount = openIterators.size() +
openTransactionalIterators.size();
+ if (openCount != 0) {
+ LOG.warn("Closing {} open iterators for store {}", openCount,
name);
for (final InMemorySessionStoreIterator it : openIterators) {
it.close();
}
+ for (final TransactionalSessionIterator it :
openTransactionalIterators) {
+ it.close();
+ }
}
endTimeMap.clear();
openIterators.clear();
+ openTransactionalIterators.clear();
open = false;
}
@@ -405,6 +520,22 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
endTimeMap.headMap(minLiveTime, false).clear();
}
+ private KeyValueIterator<Windowed<Bytes>, byte[]>
newTransactionalSessionIterator(
+ final Bytes keyFrom,
+ final Bytes keyTo,
+ final long latestSessionStartTime,
+ final long earliestSessionEndTime,
+ final long latestSessionEndTime,
+ final boolean forward) {
+ final TransactionalSessionIterator iterator = new
TransactionalSessionIterator(
+ transactionBuffer, keyFrom, keyTo, latestSessionStartTime,
+ earliestSessionEndTime, latestSessionEndTime,
+ observedStreamTime - retentionPeriod, forward,
openTransactionalIterators::remove
+ );
+ openTransactionalIterators.add(iterator);
+ return iterator;
+ }
+
private InMemorySessionStoreIterator registerNewIterator(final Bytes
keyFrom,
final Bytes keyTo,
final long
latestSessionStartTime,
@@ -423,11 +554,116 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
return iterator;
}
+ /**
+ * A session iterator backed by a transactional buffer's merge scan.
+ * Converts SessionEntryKey/byte[] pairs into Windowed<Bytes>/byte[] pairs,
+ * filtering by latestSessionStartTime.
+ */
+ private static class TransactionalSessionIterator implements
KeyValueIterator<Windowed<Bytes>, byte[]> {
+ private final
KeyValueIterator<InMemorySessionTransactionBuffer.SessionEntryKey, byte[]>
delegate;
+ private final Bytes keyFrom;
+ private final Bytes keyTo;
+ private final long latestSessionStartTime;
+ private final long oldestRetainedEndTime;
+ private final Consumer<TransactionalSessionIterator> deregister;
+ private KeyValue<Windowed<Bytes>, byte[]> prefetched;
+ private boolean closed = false;
+
+ TransactionalSessionIterator(
+ final InMemorySessionTransactionBuffer buffer,
+ final Bytes keyFrom,
+ final Bytes keyTo,
+ final long latestSessionStartTime,
+ final long earliestSessionEndTime,
+ final long latestSessionEndTime,
+ final long oldestRetainedEndTime,
+ final boolean forward,
+ final Consumer<TransactionalSessionIterator> deregister) {
+ this.keyFrom = keyFrom;
+ this.keyTo = keyTo;
+ this.latestSessionStartTime = latestSessionStartTime;
+ this.oldestRetainedEndTime = oldestRetainedEndTime;
+ this.deregister = deregister;
+
+ // startTime sorts descending, so the lower bound carries the
largest startTime and the
+ // upper bound the smallest. A null key leaves the key dimension
open (see SessionEntryKey).
+ final InMemorySessionTransactionBuffer.SessionEntryKey from =
+ new
InMemorySessionTransactionBuffer.SessionEntryKey(earliestSessionEndTime,
keyFrom, Long.MAX_VALUE);
+ final InMemorySessionTransactionBuffer.SessionEntryKey to =
+ new
InMemorySessionTransactionBuffer.SessionEntryKey(latestSessionEndTime, keyTo,
0);
+
+ this.delegate = buffer.range(from, to, forward, true);
+ }
+
+ @Override
+ public Windowed<Bytes> peekNextKey() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ return prefetched.key;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (closed) {
+ return false;
+ }
+ if (prefetched != null) {
+ return true;
+ }
+ prefetched = computeNext();
+ return prefetched != null;
+ }
+
+ @Override
+ public KeyValue<Windowed<Bytes>, byte[]> next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final KeyValue<Windowed<Bytes>, byte[]> result = prefetched;
+ prefetched = null;
+ return result;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ prefetched = null;
+ try {
+ delegate.close();
+ } finally {
+ deregister.accept(this);
+ }
+ }
+
+ private KeyValue<Windowed<Bytes>, byte[]> computeNext() {
+ while (delegate.hasNext()) {
+ final
KeyValue<InMemorySessionTransactionBuffer.SessionEntryKey, byte[]> entry =
delegate.next();
+ // The committed side is already pruned by
removeExpiredSegments, but the staged side is
+ // not; drop expired entries here. The staged scan is also
bounded only by endTime, so
+ // filter the key range and latestSessionStartTime too.
+ if (entry.key.endTime() > oldestRetainedEndTime
+ && entry.key.startTime() <= latestSessionStartTime
+ && keyInRange(entry.key.key())) {
+ final SessionWindow sessionWindow = new
SessionWindow(entry.key.startTime(), entry.key.endTime());
+ final Windowed<Bytes> windowedKey = new
Windowed<>(entry.key.key(), sessionWindow);
+ return new KeyValue<>(windowedKey, entry.value);
+ }
+ }
+ return null;
+ }
+
+ private boolean keyInRange(final Bytes key) {
+ return (keyFrom == null || key.compareTo(keyFrom) >= 0)
+ && (keyTo == null || key.compareTo(keyTo) <= 0);
+ }
+ }
+
interface ClosingCallback {
void deregisterIterator(final InMemorySessionStoreIterator iterator);
}
- private static class InMemorySessionStoreIterator implements
KeyValueIterator<Windowed<Bytes>, byte[]> {
+ static class InMemorySessionStoreIterator implements
KeyValueIterator<Windowed<Bytes>, byte[]> {
private final Iterator<Entry<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>>> endTimeIterator;
private Iterator<Entry<Bytes, ConcurrentNavigableMap<Long, byte[]>>>
keyIterator;
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
new file mode 100644
index 00000000000..00e794a7fcb
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
@@ -0,0 +1,278 @@
+/*
+ * 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.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.state.KeyValueIterator;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+/**
+ * A {@link TransactionBuffer} implementation for {@link InMemorySessionStore}.
+ * Uses a composite key of (endTime, key, startTime) to maintain correct sort
order in the staging map.
+ */
+class InMemorySessionTransactionBuffer extends
AbstractTransactionBuffer<InMemorySessionTransactionBuffer.SessionEntryKey> {
+
+ private final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeMap;
+
+ InMemorySessionTransactionBuffer(
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeMap) {
+ this.endTimeMap = endTimeMap;
+ }
+
+ /**
+ * Composite key for the session store staging map. Sorts by endTime, then
key, then startTime.
+ */
+ static final class SessionEntryKey implements Comparable<SessionEntryKey> {
+ private final long endTime;
+ private final Bytes key;
+ private final long startTime;
+
+ SessionEntryKey(final long endTime, final Bytes key, final long
startTime) {
+ this.endTime = endTime;
+ this.key = key;
+ this.startTime = startTime;
+ }
+
+ long endTime() {
+ return endTime;
+ }
+
+ Bytes key() {
+ return key;
+ }
+
+ long startTime() {
+ return startTime;
+ }
+
+ @Override
+ public int compareTo(final SessionEntryKey other) {
+ int cmp = Long.compare(this.endTime, other.endTime);
+ if (cmp != 0) {
+ return cmp;
+ }
+ if (this.key != null && other.key != null) {
+ cmp = this.key.compareTo(other.key);
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+ // Descending startTime, matching the order the reused
InMemorySessionStoreIterator emits,
+ // so the staged map and the base iterator merge in lock-step.
+ return Long.compare(other.startTime, this.startTime);
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) return true;
+ if (!(o instanceof SessionEntryKey)) return false;
+ final SessionEntryKey that = (SessionEntryKey) o;
+ return endTime == that.endTime && startTime == that.startTime &&
Objects.equals(key, that.key);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(endTime, key, startTime);
+ }
+ }
+
+ // -- Convenience methods for the store --
+
+ void stage(final Windowed<Bytes> sessionKey, final byte[] value) {
+ super.stage(new SessionEntryKey(sessionKey.window().end(),
sessionKey.key(), sessionKey.window().start()), value);
+ }
+
+ Optional<byte[]> get(final Bytes key, final long startTime, final long
endTime) {
+ return super.get(new SessionEntryKey(endTime, key, startTime));
+ }
+
+ // -- AbstractTransactionBuffer implementation --
+
+ @Override
+ int estimateKeySize(final SessionEntryKey key) {
+ return 2 * Long.BYTES + key.key().get().length;
+ }
+
+ @Override
+ void stageToBackend(final SessionEntryKey key, final byte[] value) {
+ // no-op — staging map is sufficient; no write-batch concept for
in-memory
+ }
+
+ @Override
+ ManagedKeyValueIterator<SessionEntryKey, byte[]> newBaseIterator(final
SessionEntryKey from, final SessionEntryKey to) {
+ return newBaseIterator(from, to, true, true);
+ }
+
+ @Override
+ ManagedKeyValueIterator<SessionEntryKey, byte[]> newBaseIterator(final
SessionEntryKey from, final SessionEntryKey to,
+ final
boolean forward, final boolean toInclusive) {
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> timeRange;
+ if (from != null && to != null) {
+ timeRange = endTimeMap.subMap(from.endTime(), true, to.endTime(),
true);
+ } else if (from != null) {
+ timeRange = endTimeMap.tailMap(from.endTime(), true);
+ } else if (to != null) {
+ timeRange = endTimeMap.headMap(to.endTime(), true);
+ } else {
+ timeRange = endTimeMap;
+ }
+
+ return baseIterator(forward ? timeRange : timeRange.descendingMap(),
from, to, forward);
+ }
+
+ /**
+ * Non-owner (IQ) path: eagerly deep-copies the bounded end-time range
while the caller holds
+ * the snapshot read-lock, providing true point-in-time isolation. The
returned iterator never
+ * touches the live end-time map, so concurrent owner mutation cannot
disturb it.
+ */
+ @Override
+ ManagedKeyValueIterator<SessionEntryKey, byte[]>
newBaseSnapshotIterator(final SessionEntryKey from, final SessionEntryKey to,
+
final boolean forward, final boolean toInclusive) {
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> timeRange;
+ if (from != null && to != null) {
+ timeRange = endTimeMap.subMap(from.endTime(), true, to.endTime(),
true);
+ } else if (from != null) {
+ timeRange = endTimeMap.tailMap(from.endTime(), true);
+ } else if (to != null) {
+ timeRange = endTimeMap.headMap(to.endTime(), true);
+ } else {
+ timeRange = endTimeMap;
+ }
+
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> copy = new ConcurrentSkipListMap<>();
+ for (final Map.Entry<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeEntry : timeRange.entrySet()) {
+ final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long,
byte[]>> keyCopy = new ConcurrentSkipListMap<>();
+ for (final Map.Entry<Bytes, ConcurrentNavigableMap<Long, byte[]>>
keyEntry : endTimeEntry.getValue().entrySet()) {
+ keyCopy.put(keyEntry.getKey(), new
ConcurrentSkipListMap<>(keyEntry.getValue()));
+ }
+ copy.put(endTimeEntry.getKey(), keyCopy);
+ }
+
+ return baseIterator(forward ? copy : copy.descendingMap(), from, to,
forward);
+ }
+
+ /**
+ * Builds the committed-side base iterator by reusing the non-transactional
+ * {@link InMemorySessionStore.InMemorySessionStoreIterator}, which bounds
keys at every endTime and
+ * emits in the same (endTime, key, descending-startTime) order as the
staged map. The
+ * latestSessionStartTime bound is left open here (the store's {@code
TransactionalSessionIterator}
+ * applies it); a null from/to key leaves the key dimension open.
+ */
+ private static ManagedKeyValueIterator<SessionEntryKey, byte[]>
baseIterator(
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> timeRange,
+ final SessionEntryKey from,
+ final SessionEntryKey to,
+ final boolean forward) {
+ return new SessionEntryKeyIterator(
+ new InMemorySessionStore.InMemorySessionStoreIterator(
+ from != null ? from.key() : null,
+ to != null ? to.key() : null,
+ Long.MAX_VALUE,
+ timeRange.entrySet().iterator(),
+ ignored -> { },
+ forward));
+ }
+
+ @Override
+ void flushToBase() {
+ for (final Map.Entry<SessionEntryKey, Optional<byte[]>> entry :
pendingWrites.entrySet()) {
+ final long endTime = entry.getKey().endTime();
+ final Bytes key = entry.getKey().key();
+ final long startTime = entry.getKey().startTime();
+ if (entry.getValue().isPresent()) {
+ endTimeMap.computeIfAbsent(endTime, t -> new
ConcurrentSkipListMap<>())
+ .computeIfAbsent(key, k -> new ConcurrentSkipListMap<>())
+ .put(startTime, entry.getValue().get());
+ } else {
+ final ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>> keyMap = endTimeMap.get(endTime);
+ if (keyMap != null) {
+ final ConcurrentNavigableMap<Long, byte[]> startTimeMap =
keyMap.get(key);
+ if (startTimeMap != null) {
+ startTimeMap.remove(startTime);
+ if (startTimeMap.isEmpty()) {
+ keyMap.remove(key);
+ if (keyMap.isEmpty()) {
+ endTimeMap.remove(endTime);
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ void discardPendingBatch() {
+ // no-op — no backend batch to discard
+ }
+
+ /**
+ * Adapts the non-transactional session iterator's {@code Windowed<Bytes>}
output to the
+ * {@link SessionEntryKey} space the staged-merge consumes.
+ */
+ private static final class SessionEntryKeyIterator implements
ManagedKeyValueIterator<SessionEntryKey, byte[]> {
+ private final KeyValueIterator<Windowed<Bytes>, byte[]> delegate;
+ private Runnable closeCallback;
+
+ SessionEntryKeyIterator(final KeyValueIterator<Windowed<Bytes>,
byte[]> delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return delegate.hasNext();
+ }
+
+ @Override
+ public KeyValue<SessionEntryKey, byte[]> next() {
+ final KeyValue<Windowed<Bytes>, byte[]> entry = delegate.next();
+ return new KeyValue<>(toKey(entry.key), entry.value);
+ }
+
+ @Override
+ public SessionEntryKey peekNextKey() {
+ return toKey(delegate.peekNextKey());
+ }
+
+ @Override
+ public void onClose(final Runnable closeCallback) {
+ this.closeCallback = closeCallback;
+ }
+
+ @Override
+ public void close() {
+ try {
+ delegate.close();
+ } finally {
+ if (closeCallback != null) {
+ closeCallback.run();
+ }
+ }
+ }
+
+ private static SessionEntryKey toKey(final Windowed<Bytes> windowed) {
+ return new SessionEntryKey(windowed.window().end(),
windowed.key(), windowed.window().start());
+ }
+ }
+}
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 6f5b1095f47..26f9634326b 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
@@ -49,10 +49,12 @@ import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.function.Consumer;
import static
org.apache.kafka.streams.StreamsConfig.InternalConfig.IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED;
import static
org.apache.kafka.streams.state.internals.WindowKeySchema.extractStoreKeyBytes;
@@ -72,6 +74,7 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
private final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> segmentMap = new ConcurrentSkipListMap<>();
private final Set<InMemoryWindowStoreIteratorWrapper> openIterators =
ConcurrentHashMap.newKeySet();
+ private final Set<KeyValueIterator<?, ?>> openTransactionalIterators =
ConcurrentHashMap.newKeySet();
private InternalProcessorContext<?, ?> internalProcessorContext;
private Sensor expiredRecordSensor;
@@ -81,6 +84,7 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
private volatile boolean open = false;
private final Position position;
+ private InMemoryWindowTransactionBuffer transactionBuffer;
public InMemoryWindowStore(final String name,
final long retentionPeriod,
@@ -129,23 +133,41 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
root,
(RecordBatchingStateRestoreCallback) records -> {
synchronized (position) {
+ long expiredRecords = 0;
for (final ConsumerRecord<byte[], byte[]> record :
records) {
- put(
- Bytes.wrap(extractStoreKeyBytes(record.key())),
- record.value(),
- extractStoreTimestamp(record.key())
- );
+ final Bytes key =
Bytes.wrap(extractStoreKeyBytes(record.key()));
+ final long windowStartTimestamp =
extractStoreTimestamp(record.key());
+ observedStreamTime = Math.max(observedStreamTime,
windowStartTimestamp);
+ if (windowStartTimestamp <= observedStreamTime -
retentionPeriod) {
+ expiredRecords++;
+ } else {
+ // Write directly to the committed map:
restored records are already committed.
+ putInternal(key, record.value(),
windowStartTimestamp);
+ }
ChangelogRecordDeserializationHelper.applyChecksAndUpdatePosition(
record,
consistencyEnabled,
position
);
}
+ removeExpiredSegments();
+ if (expiredRecords > 0) {
+ expiredRecordSensor.record(expiredRecords,
internalProcessorContext.currentSystemTimeMs());
+ LOG.warn("Skipping {} records for expired
segments.", expiredRecords);
+ }
}
}
);
}
open = true;
+
+ final boolean transactional = StreamsConfig.InternalConfig.getBoolean(
+ stateStoreContext.appConfigs(),
+ StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
+ false);
+ if (transactional) {
+ this.transactionBuffer = new
InMemoryWindowTransactionBuffer(segmentMap, retainDuplicates);
+ }
}
@Override
@@ -162,28 +184,41 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
if (windowStartTimestamp <= observedStreamTime - retentionPeriod) {
expiredRecordSensor.record(1.0d,
internalProcessorContext.currentSystemTimeMs());
LOG.warn("Skipping record for expired segment.");
- } else {
+ } else if (transactionBuffer != null) {
if (value != null) {
maybeUpdateSeqnumForDups();
final Bytes keyBytes = retainDuplicates ? wrapForDups(key,
seqnum) : key;
- segmentMap.computeIfAbsent(windowStartTimestamp, t -> new
ConcurrentSkipListMap<>());
- segmentMap.get(windowStartTimestamp).put(keyBytes, value);
+ transactionBuffer.stage(windowStartTimestamp, keyBytes,
value);
} else if (!retainDuplicates) {
// Skip if value is null and duplicates are allowed since
this delete is a no-op
- segmentMap.computeIfPresent(windowStartTimestamp, (t,
kvMap) -> {
- kvMap.remove(key);
- if (kvMap.isEmpty()) {
- segmentMap.remove(windowStartTimestamp);
- }
- return kvMap;
- });
+ transactionBuffer.stage(windowStartTimestamp, key, null);
}
+ } else {
+ putInternal(key, value, windowStartTimestamp);
}
StoreQueryUtils.updatePosition(position, internalProcessorContext);
}
}
+ private void putInternal(final Bytes key, final byte[] value, final long
windowStartTimestamp) {
+ if (value != null) {
+ maybeUpdateSeqnumForDups();
+ final Bytes keyBytes = retainDuplicates ? wrapForDups(key, seqnum)
: key;
+ segmentMap.computeIfAbsent(windowStartTimestamp, t -> new
ConcurrentSkipListMap<>());
+ segmentMap.get(windowStartTimestamp).put(keyBytes, value);
+ } else if (!retainDuplicates) {
+ // Skip if value is null and duplicates are allowed since this
delete is a no-op
+ segmentMap.computeIfPresent(windowStartTimestamp, (t, kvMap) -> {
+ kvMap.remove(key);
+ if (kvMap.isEmpty()) {
+ segmentMap.remove(windowStartTimestamp);
+ }
+ return kvMap;
+ });
+ }
+ }
+
@Override
public byte[] fetch(final Bytes key, final long windowStartTimestamp) {
Objects.requireNonNull(key, "key cannot be null");
@@ -194,6 +229,13 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return null;
}
+ if (transactionBuffer != null) {
+ final Optional<byte[]> staged =
transactionBuffer.get(windowStartTimestamp, key);
+ if (staged != null) {
+ return staged.orElse(null);
+ }
+ }
+
final ConcurrentNavigableMap<Bytes, byte[]> kvMap =
segmentMap.get(windowStartTimestamp);
if (kvMap == null) {
return null;
@@ -224,6 +266,15 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return WrappedInMemoryWindowStoreIterator.emptyIterator();
}
+ if (transactionBuffer != null) {
+ final Bytes keyFrom = retainDuplicates ? wrapForDups(key, 0) : key;
+ final Bytes keyTo = retainDuplicates ? wrapForDups(key,
Integer.MAX_VALUE) : key;
+ return registerTransactional(new TransactionalWindowStoreIterator(
+ transactionBuffer, keyFrom, keyTo, minTime, timeTo, forward,
retainDuplicates,
+ openTransactionalIterators::remove
+ ));
+ }
+
if (forward) {
return registerNewWindowStoreIterator(
key,
@@ -279,6 +330,14 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return KeyValueIterators.emptyIterator();
}
+ if (transactionBuffer != null) {
+ final Bytes keyFrom = (retainDuplicates && from != null) ?
wrapForDups(from, 0) : from;
+ final Bytes keyTo = (retainDuplicates && to != null) ?
wrapForDups(to, Integer.MAX_VALUE) : to;
+ return registerTransactional(new
TransactionalWindowedKeyValueIterator(
+ transactionBuffer, keyFrom, keyTo, minTime, timeTo, forward,
retainDuplicates, windowSize, openTransactionalIterators::remove
+ ));
+ }
+
if (forward) {
return registerNewWindowedKeyValueIterator(
from,
@@ -318,6 +377,12 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return KeyValueIterators.emptyIterator();
}
+ if (transactionBuffer != null) {
+ return registerTransactional(new
TransactionalWindowedKeyValueIterator(
+ transactionBuffer, null, null, minTime, timeTo, forward,
retainDuplicates, windowSize, openTransactionalIterators::remove
+ ));
+ }
+
if (forward) {
return registerNewWindowedKeyValueIterator(
null,
@@ -343,6 +408,12 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final long minTime = observedStreamTime - retentionPeriod;
+ if (transactionBuffer != null) {
+ return registerTransactional(new
TransactionalWindowedKeyValueIterator(
+ transactionBuffer, null, null, minTime + 1, Long.MAX_VALUE,
true, retainDuplicates, windowSize, openTransactionalIterators::remove
+ ));
+ }
+
return registerNewWindowedKeyValueIterator(
null,
null,
@@ -357,6 +428,12 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final long minTime = observedStreamTime - retentionPeriod;
+ if (transactionBuffer != null) {
+ return registerTransactional(new
TransactionalWindowedKeyValueIterator(
+ transactionBuffer, null, null, minTime + 1, Long.MAX_VALUE,
false, retainDuplicates, windowSize, openTransactionalIterators::remove
+ ));
+ }
+
return registerNewWindowedKeyValueIterator(
null,
null,
@@ -390,24 +467,48 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ if (transactionBuffer != null) {
+ return transactionBuffer.approximateNumUncommittedBytes();
+ }
+ return 0;
+ }
+
@Override
public void commit(final Map<TopicPartition, Long> changelogOffsets) {
- // do-nothing since it is in-memory
+ if (transactionBuffer != null) {
+ transactionBuffer.commit();
+ }
}
@Override
public void close() {
- if (openIterators.size() != 0) {
- LOG.warn("Closing {} open iterators for store {}",
openIterators.size(), name);
+ if (transactionBuffer != null) {
+ transactionBuffer.rollback();
+ }
+
+ final int openCount = openIterators.size() +
openTransactionalIterators.size();
+ if (openCount != 0) {
+ LOG.warn("Closing {} open iterators for store {}", openCount,
name);
for (final InMemoryWindowStoreIteratorWrapper it : openIterators) {
it.close();
}
+ for (final KeyValueIterator<?, ?> it : openTransactionalIterators)
{
+ it.close();
+ }
}
segmentMap.clear();
+ openTransactionalIterators.clear();
open = false;
}
+ private <T extends KeyValueIterator<?, ?>> T registerTransactional(final T
iterator) {
+ openTransactionalIterators.add(iterator);
+ return iterator;
+ }
+
long numEntries() {
return segmentMap.values().stream()
.mapToLong(Map::size)
@@ -475,6 +576,219 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return iterator;
}
+ private static Bytes lowerBoundKey(final Bytes keyFrom) {
+ // The staged composite scan needs a non-null lower bound; empty bytes
is the natural minimum.
+ return keyFrom != null ? keyFrom : Bytes.wrap(new byte[0]);
+ }
+
+ private static Bytes unwrapBound(final Bytes wrappedBound, final boolean
retainDuplicates) {
+ if (wrappedBound == null) {
+ return null;
+ }
+ return retainDuplicates ? getKey(wrappedBound) : wrappedBound;
+ }
+
+ /**
+ * Whether a stored (possibly seqnum-wrapped) key falls within the
original key range. Needed
+ * because the staged composite scan is bounded only by timestamp (and, at
the boundary
+ * timestamps, by wrapped-key order, which can admit out-of-range keys);
the committed base
+ * iterator is already correctly bounded.
+ */
+ private static boolean keyInRange(final Bytes storedKey,
+ final Bytes unwrappedFrom,
+ final Bytes unwrappedTo,
+ final boolean retainDuplicates) {
+ final Bytes key = retainDuplicates ? getKey(storedKey) : storedKey;
+ return (unwrappedFrom == null || key.compareTo(unwrappedFrom) >= 0)
+ && (unwrappedTo == null || key.compareTo(unwrappedTo) <= 0);
+ }
+
+ /**
+ * A WindowStoreIterator over the transaction buffer's merge scan,
exposing each entry's
+ * timestamp/value, filtered to the requested key.
+ */
+ private static class TransactionalWindowStoreIterator implements
WindowStoreIterator<byte[]> {
+ private final
KeyValueIterator<InMemoryWindowTransactionBuffer.WindowEntryKey, byte[]>
delegate;
+ private final Bytes unwrappedFrom;
+ private final Bytes unwrappedTo;
+ private final boolean retainDuplicates;
+ private final Consumer<KeyValueIterator<?, ?>> deregister;
+ private KeyValue<Long, byte[]> prefetched;
+ private boolean closed = false;
+
+ TransactionalWindowStoreIterator(
+ final InMemoryWindowTransactionBuffer buffer,
+ final Bytes keyFrom,
+ final Bytes keyTo,
+ final long timeFrom,
+ final long timeTo,
+ final boolean forward,
+ final boolean retainDuplicates,
+ final Consumer<KeyValueIterator<?, ?>> deregister) {
+ this.retainDuplicates = retainDuplicates;
+ this.deregister = deregister;
+ this.unwrappedFrom = unwrapBound(keyFrom, retainDuplicates);
+ this.unwrappedTo = unwrapBound(keyTo, retainDuplicates);
+ final InMemoryWindowTransactionBuffer.WindowEntryKey from =
+ new InMemoryWindowTransactionBuffer.WindowEntryKey(timeFrom,
lowerBoundKey(keyFrom));
+ final InMemoryWindowTransactionBuffer.WindowEntryKey to =
+ new InMemoryWindowTransactionBuffer.WindowEntryKey(timeTo,
keyTo);
+
+ this.delegate = buffer.range(from, to, forward, true);
+ }
+
+ @Override
+ public Long peekNextKey() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ return prefetched.key;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (closed) {
+ return false;
+ }
+ if (prefetched != null) {
+ return true;
+ }
+ prefetched = computeNext();
+ return prefetched != null;
+ }
+
+ @Override
+ public KeyValue<Long, byte[]> next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final KeyValue<Long, byte[]> result = prefetched;
+ prefetched = null;
+ return result;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ prefetched = null;
+ try {
+ delegate.close();
+ } finally {
+ deregister.accept(this);
+ }
+ }
+
+ private KeyValue<Long, byte[]> computeNext() {
+ while (delegate.hasNext()) {
+ final KeyValue<InMemoryWindowTransactionBuffer.WindowEntryKey,
byte[]> entry = delegate.next();
+ if (keyInRange(entry.key.key(), unwrappedFrom, unwrappedTo,
retainDuplicates)) {
+ return new KeyValue<>(entry.key.timestamp(), entry.value);
+ }
+ }
+ return null;
+ }
+ }
+
+ /**
+ * A Windowed KeyValueIterator over the transaction buffer's merge scan,
filtered to the key
+ * range and with the stored (possibly seqnum-wrapped) key unwrapped.
+ */
+ private static class TransactionalWindowedKeyValueIterator implements
KeyValueIterator<Windowed<Bytes>, byte[]> {
+ private final
KeyValueIterator<InMemoryWindowTransactionBuffer.WindowEntryKey, byte[]>
delegate;
+ private final Bytes unwrappedFrom;
+ private final Bytes unwrappedTo;
+ private final boolean retainDuplicates;
+ private final long windowSize;
+ private final Consumer<KeyValueIterator<?, ?>> deregister;
+ private KeyValue<Windowed<Bytes>, byte[]> prefetched;
+ private boolean closed = false;
+
+ TransactionalWindowedKeyValueIterator(
+ final InMemoryWindowTransactionBuffer buffer,
+ final Bytes keyFrom,
+ final Bytes keyTo,
+ final long timeFrom,
+ final long timeTo,
+ final boolean forward,
+ final boolean retainDuplicates,
+ final long windowSize,
+ final Consumer<KeyValueIterator<?, ?>> deregister) {
+ this.retainDuplicates = retainDuplicates;
+ this.windowSize = windowSize;
+ this.deregister = deregister;
+ this.unwrappedFrom = unwrapBound(keyFrom, retainDuplicates);
+ this.unwrappedTo = unwrapBound(keyTo, retainDuplicates);
+ final InMemoryWindowTransactionBuffer.WindowEntryKey from =
+ new InMemoryWindowTransactionBuffer.WindowEntryKey(timeFrom,
lowerBoundKey(keyFrom));
+ final InMemoryWindowTransactionBuffer.WindowEntryKey to =
+ new InMemoryWindowTransactionBuffer.WindowEntryKey(timeTo,
keyTo);
+
+ this.delegate = buffer.range(from, to, forward, true);
+ }
+
+ @Override
+ public Windowed<Bytes> peekNextKey() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ return prefetched.key;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (closed) {
+ return false;
+ }
+ if (prefetched != null) {
+ return true;
+ }
+ prefetched = computeNext();
+ return prefetched != null;
+ }
+
+ @Override
+ public KeyValue<Windowed<Bytes>, byte[]> next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final KeyValue<Windowed<Bytes>, byte[]> result = prefetched;
+ prefetched = null;
+ return result;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ prefetched = null;
+ try {
+ delegate.close();
+ } finally {
+ deregister.accept(this);
+ }
+ }
+
+ private KeyValue<Windowed<Bytes>, byte[]> computeNext() {
+ while (delegate.hasNext()) {
+ final KeyValue<InMemoryWindowTransactionBuffer.WindowEntryKey,
byte[]> entry = delegate.next();
+ if (keyInRange(entry.key.key(), unwrappedFrom, unwrappedTo,
retainDuplicates)) {
+ return new KeyValue<>(toWindowed(entry.key), entry.value);
+ }
+ }
+ return null;
+ }
+
+ private Windowed<Bytes> toWindowed(final
InMemoryWindowTransactionBuffer.WindowEntryKey entryKey) {
+ final Bytes key = retainDuplicates ? getKey(entryKey.key()) :
entryKey.key();
+ long endTime = entryKey.timestamp() + windowSize;
+ if (endTime < 0) {
+ LOG.warn("Warning: window end time was truncated to Long.MAX");
+ endTime = Long.MAX_VALUE;
+ }
+ final TimeWindow timeWindow = new TimeWindow(entryKey.timestamp(),
endTime);
+ return new Windowed<>(key, timeWindow);
+ }
+ }
+
interface ClosingCallback {
void deregisterIterator(final InMemoryWindowStoreIteratorWrapper
iterator);
@@ -689,4 +1003,59 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return new Windowed<>(key, timeWindow);
}
}
+
+ /**
+ * Yields each entry as an {@link
InMemoryWindowTransactionBuffer.WindowEntryKey} keyed by the
+ * stored (possibly seqnum-wrapped) key. Reused by {@link
InMemoryWindowTransactionBuffer} as the
+ * committed-side base iterator, so it merges in lock-step with the staged
composite map; the
+ * store's transactional read wrappers unwrap the key afterwards.
+ */
+ static final class WindowEntryKeyIterator extends
InMemoryWindowStoreIteratorWrapper
+ implements
ManagedKeyValueIterator<InMemoryWindowTransactionBuffer.WindowEntryKey, byte[]>
{
+
+ private Runnable closeCallback;
+
+ WindowEntryKeyIterator(final Bytes keyFrom,
+ final Bytes keyTo,
+ final Iterator<Map.Entry<Long,
ConcurrentNavigableMap<Bytes, byte[]>>> segmentIterator,
+ final boolean retainDuplicates,
+ final boolean forward) {
+ super(keyFrom, keyTo, segmentIterator, ignored -> { },
retainDuplicates, forward);
+ }
+
+ @Override
+ public InMemoryWindowTransactionBuffer.WindowEntryKey peekNextKey() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ return new
InMemoryWindowTransactionBuffer.WindowEntryKey(super.currentTime,
super.next.key);
+ }
+
+ @Override
+ public KeyValue<InMemoryWindowTransactionBuffer.WindowEntryKey,
byte[]> next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final KeyValue<InMemoryWindowTransactionBuffer.WindowEntryKey,
byte[]> result =
+ new KeyValue<>(new
InMemoryWindowTransactionBuffer.WindowEntryKey(super.currentTime,
super.next.key), super.next.value);
+ super.next = null;
+ return result;
+ }
+
+ @Override
+ public void onClose(final Runnable closeCallback) {
+ this.closeCallback = closeCallback;
+ }
+
+ @Override
+ public void close() {
+ try {
+ super.close();
+ } finally {
+ if (closeCallback != null) {
+ closeCallback.run();
+ }
+ }
+ }
+ }
}
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
new file mode 100644
index 00000000000..1fd2ac9f753
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
@@ -0,0 +1,203 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.common.utils.Bytes;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+/**
+ * A {@link TransactionBuffer} implementation for {@link InMemoryWindowStore}.
+ * Uses a composite key of (timestamp, key) to maintain correct sort order in
the staging map.
+ */
+class InMemoryWindowTransactionBuffer extends
AbstractTransactionBuffer<InMemoryWindowTransactionBuffer.WindowEntryKey> {
+
+ private final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> segmentMap;
+ private final boolean retainDuplicates;
+
+ InMemoryWindowTransactionBuffer(
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> segmentMap,
+ final boolean retainDuplicates) {
+ this.segmentMap = segmentMap;
+ this.retainDuplicates = retainDuplicates;
+ }
+
+ /**
+ * Composite key for the window store staging map. Sorts by timestamp
first, then by key.
+ */
+ static final class WindowEntryKey implements Comparable<WindowEntryKey> {
+ private final long timestamp;
+ private final Bytes key;
+
+ WindowEntryKey(final long timestamp, final Bytes key) {
+ this.timestamp = timestamp;
+ this.key = key;
+ }
+
+ long timestamp() {
+ return timestamp;
+ }
+
+ Bytes key() {
+ return key;
+ }
+
+ @Override
+ public int compareTo(final WindowEntryKey other) {
+ final int cmp = Long.compare(this.timestamp, other.timestamp);
+ if (cmp != 0) {
+ return cmp;
+ }
+ // A null key is an unbounded upper-bound marker used only in
range scans; it sorts after
+ // every real key at the same timestamp. (The lower bound uses
empty bytes, the natural minimum.)
+ if (this.key == null || other.key == null) {
+ return Boolean.compare(this.key == null, other.key == null);
+ }
+ return this.key.compareTo(other.key);
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) return true;
+ if (!(o instanceof WindowEntryKey)) return false;
+ final WindowEntryKey that = (WindowEntryKey) o;
+ return timestamp == that.timestamp && Objects.equals(key,
that.key);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(timestamp, key);
+ }
+ }
+
+ // -- Convenience methods for the store --
+
+ void stage(final long timestamp, final Bytes key, final byte[] value) {
+ super.stage(new WindowEntryKey(timestamp, key), value);
+ }
+
+ Optional<byte[]> get(final long timestamp, final Bytes key) {
+ return super.get(new WindowEntryKey(timestamp, key));
+ }
+
+ // -- AbstractTransactionBuffer implementation --
+
+ @Override
+ int estimateKeySize(final WindowEntryKey key) {
+ return Long.BYTES + key.key().get().length;
+ }
+
+ @Override
+ void stageToBackend(final WindowEntryKey key, final byte[] value) {
+ // no-op — staging map is sufficient; no write-batch concept for
in-memory
+ }
+
+ @Override
+ ManagedKeyValueIterator<WindowEntryKey, byte[]> newBaseIterator(final
WindowEntryKey from, final WindowEntryKey to) {
+ return newBaseIterator(from, to, true, true);
+ }
+
+ @Override
+ ManagedKeyValueIterator<WindowEntryKey, byte[]> newBaseIterator(final
WindowEntryKey from, final WindowEntryKey to,
+ final
boolean forward, final boolean toInclusive) {
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> timeRange;
+ if (from != null && to != null) {
+ timeRange = segmentMap.subMap(from.timestamp(), true,
to.timestamp(), true);
+ } else if (from != null) {
+ timeRange = segmentMap.tailMap(from.timestamp(), true);
+ } else if (to != null) {
+ timeRange = segmentMap.headMap(to.timestamp(), true);
+ } else {
+ timeRange = segmentMap;
+ }
+
+ return baseIterator(forward ? timeRange : timeRange.descendingMap(),
from, to, forward);
+ }
+
+ /**
+ * Non-owner (IQ) path: eagerly deep-copies the bounded time range while
the caller holds the
+ * snapshot read-lock, providing true point-in-time isolation. The
returned iterator never
+ * touches the live segment map, so concurrent owner mutation cannot
disturb it.
+ */
+ @Override
+ ManagedKeyValueIterator<WindowEntryKey, byte[]>
newBaseSnapshotIterator(final WindowEntryKey from, final WindowEntryKey to,
+
final boolean forward, final boolean toInclusive) {
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> timeRange;
+ if (from != null && to != null) {
+ timeRange = segmentMap.subMap(from.timestamp(), true,
to.timestamp(), true);
+ } else if (from != null) {
+ timeRange = segmentMap.tailMap(from.timestamp(), true);
+ } else if (to != null) {
+ timeRange = segmentMap.headMap(to.timestamp(), true);
+ } else {
+ timeRange = segmentMap;
+ }
+
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> copy = new ConcurrentSkipListMap<>();
+ for (final Map.Entry<Long, ConcurrentNavigableMap<Bytes, byte[]>>
segment : timeRange.entrySet()) {
+ copy.put(segment.getKey(), new
ConcurrentSkipListMap<>(segment.getValue()));
+ }
+
+ return baseIterator(forward ? copy : copy.descendingMap(), from, to,
forward);
+ }
+
+ /**
+ * Builds the committed-side base iterator by reusing the
non-transactional segment iterator
+ * ({@link InMemoryWindowStore.WindowEntryKeyIterator}), which bounds keys
at every timestamp and
+ * emits in the same (timestamp, stored-key) order as the staged map. A
null from/to key (and the
+ * empty-bytes lower-bound placeholder) leaves that side of the key
dimension open.
+ */
+ private ManagedKeyValueIterator<WindowEntryKey, byte[]> baseIterator(
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> timeRange,
+ final WindowEntryKey from,
+ final WindowEntryKey to,
+ final boolean forward) {
+ final Bytes keyFrom = (from != null && from.key() != null &&
from.key().get().length > 0) ? from.key() : null;
+ final Bytes keyTo = (to != null) ? to.key() : null;
+ return new InMemoryWindowStore.WindowEntryKeyIterator(
+ keyFrom, keyTo, timeRange.entrySet().iterator(), retainDuplicates,
forward);
+ }
+
+ @Override
+ void flushToBase() {
+ for (final Map.Entry<WindowEntryKey, Optional<byte[]>> entry :
pendingWrites.entrySet()) {
+ final long ts = entry.getKey().timestamp();
+ final Bytes key = entry.getKey().key();
+ if (entry.getValue().isPresent()) {
+ segmentMap.computeIfAbsent(ts, t -> new
ConcurrentSkipListMap<>()).put(key, entry.getValue().get());
+ } else {
+ final ConcurrentNavigableMap<Bytes, byte[]> kvMap =
segmentMap.get(ts);
+ if (kvMap != null) {
+ kvMap.remove(key);
+ if (kvMap.isEmpty()) {
+ segmentMap.remove(ts);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ void discardPendingBatch() {
+ // no-op — no backend batch to discard
+ }
+
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java
index 39a1970f533..0bc97b29729 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java
@@ -179,10 +179,19 @@ public abstract class AbstractSessionBytesStoreTest {
abstract StoreType storeType();
+ /** Overridden by subclasses to exercise the transactional (staged-write)
code path. */
+ boolean transactional() {
+ return false;
+ }
+
@BeforeEach
public void setUp() {
sessionStore = buildSessionStore(RETENTION_PERIOD, Serdes.String(),
Serdes.Long());
recordCollector = new MockRecordCollector();
+ final Properties streamsConfig = StreamsTestUtils.getStreamsConfig();
+ if (transactional()) {
+ streamsConfig.put(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
true);
+ }
context = new InternalMockProcessorContext<>(
TestUtils.tempDirectory(),
Serdes.String(),
@@ -191,7 +200,8 @@ public abstract class AbstractSessionBytesStoreTest {
new ThreadCache(
new LogContext("testCache"),
0,
- new MockStreamsMetrics(new Metrics())));
+ new MockStreamsMetrics(new Metrics())),
+ new StreamsConfig(streamsConfig));
context.setTime(1L);
sessionStore.init(context, sessionStore);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractWindowBytesStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractWindowBytesStoreTest.java
index c5e9c5b59fd..edcb73bad02 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractWindowBytesStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractWindowBytesStoreTest.java
@@ -99,12 +99,21 @@ public abstract class AbstractWindowBytesStoreTest {
final boolean
retainDuplicates,
final Serde<K> keySerde,
final Serde<V>
valueSerde);
+ /** Overridden by subclasses to exercise the transactional (staged-write)
code path. */
+ boolean transactional() {
+ return false;
+ }
+
@BeforeEach
protected void setup() {
-
+
windowStore = buildWindowStore(RETENTION_PERIOD, WINDOW_SIZE, false,
Serdes.Integer(), Serdes.String());
recordCollector = new MockRecordCollector();
+ final Properties streamsConfig = StreamsTestUtils.getStreamsConfig();
+ if (transactional()) {
+ streamsConfig.put(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
true);
+ }
context = new InternalMockProcessorContext<>(
baseDir,
Serdes.String(),
@@ -113,7 +122,8 @@ public abstract class AbstractWindowBytesStoreTest {
new ThreadCache(
new LogContext("testCache"),
0,
- new MockStreamsMetrics(new Metrics())));
+ new MockStreamsMetrics(new Metrics())),
+ new StreamsConfig(streamsConfig));
context.setTime(1L);
windowStore.init(context, windowStore);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTransactionalSessionStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTransactionalSessionStoreTest.java
new file mode 100644
index 00000000000..260b6e08d2e
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTransactionalSessionStoreTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Set;
+
+import static org.apache.kafka.test.StreamsTestUtils.valuesToSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Runs the full {@link AbstractSessionBytesStoreTest} suite against the
in-memory session store with
+ * transactional state stores enabled, so the staged-write/merge path is held
to the same behavioural
+ * contract as the non-transactional store. Adds transactional-specific cases
that the base suite (which
+ * only ever reads staged, un-committed writes) does not cover.
+ */
+public class InMemoryTransactionalSessionStoreTest extends
AbstractSessionBytesStoreTest {
+
+ @Override
+ StoreType storeType() {
+ return StoreType.InMemoryStore;
+ }
+
+ @Override
+ boolean transactional() {
+ return true;
+ }
+
+ @Test
+ public void shouldReadAcrossCommitBoundaries() {
+ final Windowed<String> a0 = new Windowed<>("a", new SessionWindow(0,
0));
+ final Windowed<String> a1 = new Windowed<>("a", new SessionWindow(10,
20));
+
+ sessionStore.put(a0, 1L); // staged
+ sessionStore.commit(Collections.emptyMap()); // -> committed
+ sessionStore.put(a1, 2L); // staged on top of
committed
+
+ // committed (a0) and staged (a1) are both visible before commit
+ try (final KeyValueIterator<Windowed<String>, Long> it =
sessionStore.fetch("a")) {
+ assertEquals(Set.of(1L, 2L), valuesToSet(it));
+ }
+ sessionStore.commit(Collections.emptyMap()); // a1 -> committed;
still visible
+ try (final KeyValueIterator<Windowed<String>, Long> it =
sessionStore.fetch("a")) {
+ assertEquals(Set.of(1L, 2L), valuesToSet(it));
+ }
+
+ // a staged tombstone hides the committed session, and stays hidden
after commit
+ sessionStore.remove(a0);
+ try (final KeyValueIterator<Windowed<String>, Long> it =
sessionStore.fetch("a")) {
+ assertEquals(Set.of(2L), valuesToSet(it));
+ }
+ sessionStore.commit(Collections.emptyMap());
+ try (final KeyValueIterator<Windowed<String>, Long> it =
sessionStore.fetch("a")) {
+ assertEquals(Set.of(2L), valuesToSet(it));
+ }
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTransactionalWindowStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTransactionalWindowStoreTest.java
new file mode 100644
index 00000000000..5660dbebae0
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryTransactionalWindowStoreTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.streams.state.WindowStore;
+import org.apache.kafka.streams.state.WindowStoreIterator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static java.time.Duration.ofMillis;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * Runs the full {@link AbstractWindowBytesStoreTest} suite against the
in-memory window store with
+ * transactional state stores enabled, so the staged-write/merge path is held
to the same behavioural
+ * contract as the non-transactional store.
+ */
+public class InMemoryTransactionalWindowStoreTest extends
AbstractWindowBytesStoreTest {
+
+ private static final String STORE_NAME =
"InMemoryTransactionalWindowStore";
+
+ @Override
+ <K, V> WindowStore<K, V> buildWindowStore(final long retentionPeriod,
+ final long windowSize,
+ final boolean retainDuplicates,
+ final Serde<K> keySerde,
+ final Serde<V> valueSerde) {
+ return Stores.windowStoreBuilder(
+ Stores.inMemoryWindowStore(
+ STORE_NAME,
+ ofMillis(retentionPeriod),
+ ofMillis(windowSize),
+ retainDuplicates),
+ keySerde,
+ valueSerde)
+ .build();
+ }
+
+ @Override
+ boolean transactional() {
+ return true;
+ }
+
+ /**
+ * Transactional reads are snapshot-isolated: an open iterator reflects
the staged state as of when
+ * it was opened and does not observe records staged afterwards (the
non-transactional store iterates
+ * the live map and does observe them). It must still not throw. This
overrides the base test's
+ * live-iteration expectation accordingly.
+ */
+ @Test
+ @Override
+ public void shouldNotThrowConcurrentModificationException() {
+ long currentTime = 0;
+ windowStore.put(1, "one", currentTime);
+
+ currentTime += WINDOW_SIZE * 10;
+ windowStore.put(1, "two", currentTime);
+
+ try (final KeyValueIterator<Windowed<Integer>, String> iterator =
windowStore.all()) {
+ currentTime += WINDOW_SIZE * 10;
+ windowStore.put(1, "three", currentTime);
+
+ currentTime += WINDOW_SIZE * 10;
+ windowStore.put(2, "four", currentTime);
+
+ assertEquals(windowedPair(1, "one", 0), iterator.next());
+ assertEquals(windowedPair(1, "two", WINDOW_SIZE * 10),
iterator.next());
+ assertFalse(iterator.hasNext());
+ }
+ }
+
+ @Test
+ public void shouldReadAcrossCommitBoundaries() {
+ windowStore.put(1, "one", 0); // staged
+ windowStore.commit(Collections.emptyMap()); // -> committed
+ windowStore.put(1, "two", WINDOW_SIZE * 10); // staged on
top of committed
+
+ // committed (one) and staged (two) are both visible before commit,
and after
+ try (final WindowStoreIterator<String> it = windowStore.fetch(1, 0,
WINDOW_SIZE * 10)) {
+ assertEquals(new KeyValue<>(0L, "one"), it.next());
+ assertEquals(new KeyValue<>(WINDOW_SIZE * 10, "two"), it.next());
+ assertFalse(it.hasNext());
+ }
+ windowStore.commit(Collections.emptyMap());
+ try (final WindowStoreIterator<String> it = windowStore.fetch(1, 0,
WINDOW_SIZE * 10)) {
+ assertEquals(new KeyValue<>(0L, "one"), it.next());
+ assertEquals(new KeyValue<>(WINDOW_SIZE * 10, "two"), it.next());
+ assertFalse(it.hasNext());
+ }
+
+ // a staged tombstone hides the committed record, and stays hidden
after commit
+ windowStore.put(1, null, 0);
+ try (final WindowStoreIterator<String> it = windowStore.fetch(1, 0,
WINDOW_SIZE * 10)) {
+ assertEquals(new KeyValue<>(WINDOW_SIZE * 10, "two"), it.next());
+ assertFalse(it.hasNext());
+ }
+ windowStore.commit(Collections.emptyMap());
+ try (final WindowStoreIterator<String> it = windowStore.fetch(1, 0,
WINDOW_SIZE * 10)) {
+ assertEquals(new KeyValue<>(WINDOW_SIZE * 10, "two"), it.next());
+ assertFalse(it.hasNext());
+ }
+ }
+}
