Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5691#discussion_r174385360
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java
---
@@ -40,6 +54,81 @@
*/
public class KvStateRegistryTest extends TestLogger {
+ @Test
+ public void testKvStateEntry() throws InterruptedException {
+ final int threads = 10;
+
+ final CountDownLatch latch1 = new CountDownLatch(threads);
+ final CountDownLatch latch2 = new CountDownLatch(1);
+
+ final List<KvStateInfo<?, ?, ?>> infos =
Collections.synchronizedList(new ArrayList<>());
+
+ final JobID jobID = new JobID();
+
+ final JobVertexID jobVertexId = new JobVertexID();
+ final KeyGroupRange keyGroupRange = new KeyGroupRange(0, 1);
+ final String registrationName = "foobar";
+
+ final KvStateRegistry kvStateRegistry = new KvStateRegistry();
+ final KvStateID stateID = kvStateRegistry.registerKvState(
+ jobID,
+ jobVertexId,
+ keyGroupRange,
+ registrationName,
+ new DummyKvState()
+ );
+
+ for (int i = 0; i < threads; i++) {
+ new Thread(() -> {
+ final KvStateEntry<?, ?, ?> kvState =
kvStateRegistry.getKvState(stateID);
+ final KvStateInfo<?, ?, ?> stateInfo =
kvState.getInfoForCurrentThread();
+ infos.add(stateInfo);
+
+ latch1.countDown();
+ try {
+ latch2.await();
+ } catch (InterruptedException e) {
+ Assert.fail(e.getMessage());
--- End diff --
this would never directly fail the test and we may loose the exception
---
