[GitHub] [flink] SmirAlex commented on a diff in pull request #20919: [FLINK-29405] Fix unstable test InputFormatCacheLoaderTest

[GitBox <gi...@apache.org>]

SmirAlex commented on code in PR #20919:
URL: https://github.com/apache/flink/pull/20919#discussion_r1036683826


##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/table/fullcache/inputformat/InputFormatCacheLoaderTest.java:
##########
@@ -149,19 +155,33 @@ void testCloseAndInterruptDuringReload() throws Exception {
         Future<?> future = executorService.submit(cacheLoader);
         executorService.shutdownNow(); // internally interrupts a thread
         assertThatNoException().isThrownBy(future::get); // wait for the end
-        // check that we didn't process all elements, but reacted on interruption
-        assertThat(sleepCounter).hasValueLessThan(totalSleepCount);
         assertThat(metricGroup.numLoadFailuresCounter.getCount()).isEqualTo(0);
+    }
 
-        sleepCounter.set(0);
+    @Test
+    void testCloseDuringReload() throws Exception {
+        AtomicInteger recordsCounter = new AtomicInteger(0);
+        int totalRecords = TestCacheLoader.DATA.size() + 1; // 1 key with 2 records
+        CountDownLatch latch = new CountDownLatch(1);
+        Runnable reloadAction =
+                ThrowingRunnable.unchecked(
+                        () -> {
+                            recordsCounter.incrementAndGet();
+                            latch.await();
+                        });
+        InputFormatCacheLoader cacheLoader = createCacheLoader(0, reloadAction);
+        InterceptingCacheMetricGroup metricGroup = new InterceptingCacheMetricGroup();
+        cacheLoader.open(metricGroup);
 
         // check closing
-        executorService = Executors.newSingleThreadExecutor();
-        future = executorService.submit(cacheLoader);
+        ExecutorService executorService = Executors.newSingleThreadExecutor();
+        Future<?> future = executorService.submit(cacheLoader);
         cacheLoader.close();
-        assertThatNoException().isThrownBy(future::get); // wait for the end
+        latch.countDown();
+        future.get(); // wait for the end
+        executorService.shutdown();

Review Comment:
   But do we need to wait here? Currently there are 2 ways to stop cache loading: via `CacheLoader#close()` method and via interrupting a thread. This one test only first way of stopping. Actually, we could limit ourselves to the first way only, but I decided to also implement interrupt handling for completeness and, probably, more fast reaction on closing.



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org