nicktelford commented on code in PR #22626:
URL: https://github.com/apache/kafka/pull/22626#discussion_r3452095134
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:
##########
Review Comment:
Good catch, but I've decided to go in a different direction: taking a
snapshot of the range being iterated when an IQ thread constructs an iterator.
This retains the `TreeMap` for the non-transactional path (and it's clear
from KAFKA-8802 that ConcurrentSkipListMap's performance was an issue in the
past), and keeps non-transactional behaviour completely untouched.
It's important to remember that the existing, non-transactional path
unconditionally takes a snapshot of the range of keys in the map being
iterated, but then produces the live value for each key during iteration. This
prevents ConcurrentModificationException, and ensures that a StreamThread that
modifies keys it iterates sees its own writes.
When transactions are enabled, we have a few different paths:
- `StreamThread`: the owner thread iterates the `TreeMap` *directly*. No
snapshot needed. We can do this because writes target the transaction buffer,
not the store, so it sees its own writes via the
`AbstractTransactionBuffer#StagedMergeIterator`. It's not possible for the
store itself to be updated during iteration because you can't `commit` from
within a `Processor`.
- IQ threads: takes a snapshot of the range being iterated (including
values!) to provide snapshot isolation; this snapshot is done under the same
`readLock` that's used to snapshot the transaction buffer. Compared with the
non-transactional snapshots, the performance impact should be similar,
including memory usage (because it's only the value references being copied,
not the values themselves). The main caveat is that `byte[]` values are
effectively "pinned" on the heap until the `Iterator` is closed, preventing GC.
One concern I have: as I discovered in KAFKA-8802, performance of
`ConcurrentSkipListMap` was deemed a regression vs. `TreeMap`, but our
`AbstractTransactionBuffer#pendingWrites` uses a `ConcurrentSkipListMap`, which
will likely impact the performance of in-memory stores. I'm going to
investigate if there's a way to optimize this a bit better, but I think that's
orthogonal to this PR.
--
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]
