gabriellefu commented on code in PR #23184:
URL: https://github.com/apache/kafka/pull/23184#discussion_r3897706968
##########
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:
thank you for pointing it cout, have drop the iterator
--
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]
