frankvicky commented on code in PR #23184:
URL: https://github.com/apache/kafka/pull/23184#discussion_r3864546469
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java:
##########
@@ -128,29 +128,35 @@ public static <R> QueryResult<R> handleBasicQueries(
final QueryResult<R> result;
final QueryHandler<?> handler =
QUERY_HANDLER_MAP.get(query.getClass());
- synchronized (position) {
- if (handler == null) {
- result = QueryResult.forUnknownQueryType(query, store);
- } else if (context == null || !isPermitted(position,
positionBound, context.taskId().partition())) {
- result = QueryResult.notUpToBound(
- position,
- positionBound,
- context == null ? null : context.taskId().partition()
- );
- } else {
- result = ((QueryHandler<R>) handler).apply(
- query,
- positionBound,
- config,
- store
- );
- }
- if (config.isCollectExecutionInfo()) {
- result.addExecutionInfo(
- "Handled in " + store.getClass() + " in " +
(System.nanoTime() - start) + "ns"
- );
+ // Take the store monitor before the position lock, matching the write
paths, so a
+ // concurrent put/query cannot deadlock (KAFKA-19629). Callers must
not hold the
+ // position lock when calling in, unless nothing else locks their
store monitor
Review Comment:
This caller contract is only documented here, not enforced:
`InMemoryWindowStore` and `InMemorySessionStore` still call in while holding
the position lock (position → store order), which is safe today only because
nothing else locks their monitors. If someone later adds a `synchronized`
method to either store, the deadlock silently comes back.
Could we add a matching warning comment on those two stores' `query()`
methods, so the invariant is visible at the point where it could be broken?
(Alternatively, restoring the outer store lock there would make the contract
exception-free — it was harmless.)
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java:
##########
@@ -638,6 +638,65 @@ public void
shouldReportUncommittedPositionForTransactionalStore() {
}
}
+ @Test
+ public void shouldNotDeadlockOnConcurrentPutAndQuery() throws Exception {
+ // KAFKA-19629: put() takes the store monitor and then the position
lock, so IQ queries
+ // must take the two locks in the same order.
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig())
+ );
+ final InMemoryKeyValueStore store = new
InMemoryKeyValueStore("concurrency-store");
+ store.init(ctx, store);
+ ctx.setRecordContext(new ProcessorRecordContext(0, 1, 0, "topic", new
RecordHeaders()));
+
+ final int iterations = 5000;
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ final Thread writer = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ store.put(bytesKey("key" + (i % 100)), bytesValue("value"
+ i));
+ }
+ } catch (final Throwable t) {
+ failure.set(t);
+ }
+ }, "writer");
+
+ final Thread reader = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ final
org.apache.kafka.streams.query.QueryResult<KeyValueIterator<Bytes, byte[]>>
result = store.query(
+
org.apache.kafka.streams.query.RangeQuery.withNoBounds(),
+
org.apache.kafka.streams.query.PositionBound.unbounded(),
+ new org.apache.kafka.streams.query.QueryConfig(false));
+ try (KeyValueIterator<Bytes, byte[]> iterator =
result.getResult()) {
Review Comment:
Consuming the iterator here races the unsynchronized live-`TreeMap` reads
inside `InMemoryKeyValueIterator#hasNext()` (`map.containsKey` without the
store lock). That's pre-existing store behavior, not something this PR
introduces, but it could make this test rarely flaky. The deadlock repro only
needs the locks taken in `query()` itself — opening and closing the iterator
(without `hasNext()`/`next()`) is enough.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java:
##########
@@ -638,6 +638,65 @@ public void
shouldReportUncommittedPositionForTransactionalStore() {
}
}
+ @Test
+ public void shouldNotDeadlockOnConcurrentPutAndQuery() throws Exception {
+ // KAFKA-19629: put() takes the store monitor and then the position
lock, so IQ queries
+ // must take the two locks in the same order.
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig())
+ );
+ final InMemoryKeyValueStore store = new
InMemoryKeyValueStore("concurrency-store");
+ store.init(ctx, store);
+ ctx.setRecordContext(new ProcessorRecordContext(0, 1, 0, "topic", new
RecordHeaders()));
+
+ final int iterations = 5000;
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ final Thread writer = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ store.put(bytesKey("key" + (i % 100)), bytesValue("value"
+ i));
+ }
+ } catch (final Throwable t) {
+ failure.set(t);
+ }
+ }, "writer");
+
+ final Thread reader = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ final
org.apache.kafka.streams.query.QueryResult<KeyValueIterator<Bytes, byte[]>>
result = store.query(
+
org.apache.kafka.streams.query.RangeQuery.withNoBounds(),
Review Comment:
nit: the RocksDB and LRU variants of this test import
`RangeQuery`/`PositionBound`/`QueryConfig`/`QueryResult` — could we do the same
here instead of fully-qualified names, for consistency across the three new
tests?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryLRUCacheStoreTest.java:
##########
@@ -162,4 +176,63 @@ public void testRestoreEvict() {
// and there are no other entries ...
assertEquals(10, driver.sizeOf(store));
}
+
+ @Test
+ public void shouldNotDeadlockOnConcurrentPutAndQuery() throws Exception {
+ // KAFKA-19629: put() takes the store monitor and then the position
lock, so IQ queries
+ // must take the two locks in the same order.
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig()));
+ final MemoryNavigableLRUCache cache = new
MemoryNavigableLRUCache("lru-put-store", 100);
+ cache.init((StateStoreContext) ctx, cache);
+
+ final int iterations = 5000;
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ final Thread writer = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ cache.put(
+ Bytes.wrap(("key" + (i % 100)).getBytes()),
+ ("value" + i).getBytes());
+ }
+ } catch (final Throwable t) {
+ failure.set(t);
+ }
+ }, "writer");
+
+ final Thread reader = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> result
= cache.query(
+ RangeQuery.withNoBounds(),
+ PositionBound.unbounded(),
+ new QueryConfig(false));
+ try (KeyValueIterator<Bytes, byte[]> iterator =
result.getResult()) {
+ if (iterator.hasNext()) {
+ iterator.next();
+ }
+ }
+ }
+ } catch (final Throwable t) {
+ failure.set(t);
+ }
+ }, "writer-query-reader");
Review Comment:
nit: `"writer-query-reader"` → `"iq-reader"`, to match the other two tests?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java:
##########
@@ -1802,6 +1808,63 @@ public void putAllShouldThrowOnClosedStore() {
stringSerializer.serialize(null, "v1")))));
}
+ @Test
+ public void shouldNotDeadlockOnConcurrentPutAndQuery() throws Exception {
+ // KAFKA-19629: put() takes the store monitor and then the position
lock, so IQ queries
+ // must take the two locks in the same order.
+ rocksDBStore.init(context, rocksDBStore);
+
+ final int iterations = 5000;
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ final Thread writer = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ rocksDBStore.put(
+ new Bytes(stringSerializer.serialize(null, "key" + (i
% 100))),
+ stringSerializer.serialize(null, "value" + i));
+ }
+ } catch (final Throwable t) {
+ failure.set(t);
+ }
+ }, "writer");
+
+ final Thread reader = new Thread(() -> {
+ try {
+ for (int i = 0; i < iterations; i++) {
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> result
= rocksDBStore.query(
+ RangeQuery.withNoBounds(),
+ PositionBound.unbounded(),
+ new QueryConfig(false));
+ try (KeyValueIterator<Bytes, byte[]> iterator =
result.getResult()) {
+ if (iterator.hasNext()) {
+ iterator.next();
+ }
+ }
+ }
+ } catch (final Throwable t) {
+ failure.set(t);
+ }
+ }, "iq-reader");
+
+ writer.setDaemon(true);
+ reader.setDaemon(true);
+ writer.start();
+ reader.start();
+ writer.join(TimeUnit.SECONDS.toMillis(60));
+ reader.join(TimeUnit.SECONDS.toMillis(60));
Review Comment:
nit: on a deadlock these two joins wait sequentially, so the test takes up
to 120s to fail. Sharing one deadline across both joins (or a JUnit `@Timeout`)
would halve that. Same applies to the other two tests.
--
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]
