bbejeck commented on code in PR #22681:
URL: https://github.com/apache/kafka/pull/22681#discussion_r3493287714
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:
##########
@@ -267,6 +270,104 @@ public long approximateNumEntries() {
return map.size();
}
+ @Override
+ public ReadOnlyKeyValueStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ final boolean consultBuffer = isolationLevel ==
IsolationLevel.READ_UNCOMMITTED;
+ return new ReadOnlyView(consultBuffer);
+ }
+
+ /**
+ * Read-only view of this store bound to an isolation level. When {@code
consultBuffer}
+ * is true, reads go through the transaction buffer if present
(READ_UNCOMMITTED);
+ * otherwise they read the base {@code map} directly (READ_COMMITTED, or
any level when
+ * the store is non-transactional).
+ */
+ private final class ReadOnlyView implements ReadOnlyKeyValueStore<Bytes,
byte[]> {
+
+ private final boolean consultBuffer;
+
+ ReadOnlyView(final boolean consultBuffer) {
+ this.consultBuffer = consultBuffer;
+ }
+
+ @Override
+ public byte[] get(final Bytes key) {
+ Objects.requireNonNull(key, "key cannot be null");
+ synchronized (InMemoryKeyValueStore.this) {
Review Comment:
synchonizing here on `InMemoryKeyValueStore.this` can result in a
`ConcurrentModificationException` as the `commit()` is guarded by
`snapshotLock.writeLock` - I think this need and other spots like it need to
use `snapshotLock.readLock`
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:
##########
@@ -267,6 +270,104 @@ public long approximateNumEntries() {
return map.size();
}
+ @Override
+ public ReadOnlyKeyValueStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ final boolean consultBuffer = isolationLevel ==
IsolationLevel.READ_UNCOMMITTED;
+ return new ReadOnlyView(consultBuffer);
+ }
+
+ /**
+ * Read-only view of this store bound to an isolation level. When {@code
consultBuffer}
+ * is true, reads go through the transaction buffer if present
(READ_UNCOMMITTED);
+ * otherwise they read the base {@code map} directly (READ_COMMITTED, or
any level when
+ * the store is non-transactional).
+ */
+ private final class ReadOnlyView implements ReadOnlyKeyValueStore<Bytes,
byte[]> {
+
+ private final boolean consultBuffer;
+
+ ReadOnlyView(final boolean consultBuffer) {
+ this.consultBuffer = consultBuffer;
+ }
+
+ @Override
+ public byte[] get(final Bytes key) {
+ Objects.requireNonNull(key, "key cannot be null");
+ synchronized (InMemoryKeyValueStore.this) {
+ if (consultBuffer && transactionBuffer != null) {
+ final java.util.Optional<byte[]> staged =
transactionBuffer.get(key);
+ if (staged != null) {
+ return staged.orElse(null);
+ }
+ }
+ return map.get(key);
+ }
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> range(final Bytes from, final
Bytes to) {
+ return doRange(from, to, true);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> reverseRange(final Bytes from,
final Bytes to) {
+ return doRange(from, to, false);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> all() {
+ return doRange(null, null, true);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> reverseAll() {
+ return doRange(null, null, false);
+ }
+
+ private KeyValueIterator<Bytes, byte[]> doRange(final Bytes from,
final Bytes to, final boolean forward) {
+ synchronized (InMemoryKeyValueStore.this) {
Review Comment:
same issue here - I think we solved this issue in a previous PR the buffer's
non-owner/IQ read path (`AbstractTransactionBuffer.snapshotScan` →
`newBaseSnapshotIterator`) eagerly copies the bounded base range into a TreeMap
while holding snapshotLock.readLock(), and commit()/rollback() take
`snapshotLock.writeLock()` before touching the base map.
--
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]
