nicktelford commented on code in PR #22652:
URL: https://github.com/apache/kafka/pull/22652#discussion_r3476435149
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java:
##########
@@ -423,6 +525,86 @@ private InMemorySessionStoreIterator
registerNewIterator(final Bytes keyFrom,
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 long latestSessionStartTime;
+ private KeyValue<Windowed<Bytes>, byte[]> prefetched;
+
+ TransactionalSessionIterator(
+ final InMemorySessionTransactionBuffer buffer,
+ final Bytes keyFrom,
+ final Bytes keyTo,
+ final long latestSessionStartTime,
+ final long earliestSessionEndTime,
+ final long latestSessionEndTime,
+ final boolean forward) {
+ this.latestSessionStartTime = latestSessionStartTime;
+
+ final InMemorySessionTransactionBuffer.SessionEntryKey from =
+ new InMemorySessionTransactionBuffer.SessionEntryKey(
+ earliestSessionEndTime,
+ keyFrom != null ? keyFrom : Bytes.wrap(new byte[0]),
+ 0);
+ final InMemorySessionTransactionBuffer.SessionEntryKey to =
+ new InMemorySessionTransactionBuffer.SessionEntryKey(
+ latestSessionEndTime,
+ keyTo != null ? keyTo : Bytes.wrap(new byte[]{(byte) 0xFF,
(byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF,
(byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF,
(byte) 0xFF, (byte) 0xFF, (byte) 0xFF}),
Review Comment:
Fixed in 2c90ce6b319468a80c298ec39ebc9d3d72caf87a
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
