gabriellefu commented on code in PR #23184:
URL: https://github.com/apache/kafka/pull/23184#discussion_r3897731636
##########
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:
sounds good, have changed
--
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]
