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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
