SmirAlex commented on code in PR #20919:
URL: https://github.com/apache/flink/pull/20919#discussion_r1041842260
##########
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:
Agree, now all tests on `InputFormatCacheLoader` uses try-with-resources, so
all threads used for cache loading should be closed after test passes
--
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]
