[common] Ensure ParallelExecution waits for reader cleanup

[QuakeWang]

Purpose

ParallelExecution marks tasks complete before closing the underlying readers, while close() previously only interrupted the executor. As a result, close() could return while reader cleanup was still running, leaving input streams temporarily
open.

Wait for executor termination in close() so reader resources are released before it returns. Add deterministic coverage for the race and remove the sleep workaround from IndexBootstrapTest.

Tests

• mvn -pl paimon-core -am -DfailIfNoTests=false -DwildcardSuites=none -Dtest=ParallelExecutionTest,IndexBootstrapTest test
• Ran testBootstrapIgnoresQueryAuth 1,000 times concurrently on JDK 11

[JingsongLi]

Could we avoid interrupting workers that have already finished reading and are only closing their readers?

latch.countDown() happens before the try-with-resources closes the iterator. Therefore, after take() observes completion, the worker may still be inside RecordReader.close(). Calling shutdownNow() here interrupts that cleanup. awaitTermination() then only guarantees that the worker exited; it does not guarantee that cleanup completed successfully.

I reproduced this with a reader whose close() waits on an interruptible semaphore. ParallelExecution.close() interrupted the cleanup and returned before the semaphore was released, with the cleanup operation never completed. The new test uses acquireUninterruptibly(), so it cannot expose this race.

For the normal completed-read path, could we use executorService.shutdown() when latch.getCount() == 0, and reserve shutdownNow() for early cancellation? Alternatively, moving the countdown after resource cleanup would also avoid this race.

[QuakeWang]

Thanks @JingsongLi for catching this. You are right that shutdownNow() could interrupt reader cleanup after take() had already observed completion.

Fixed in d16774b2b. ParallelExecution.close() now uses shutdown() when all readers have finished reading, while retaining shutdownNow() for early cancellation. I chose this approach to preserve the existing behavior where take() does not wait for reader cleanup.

I also changed the test to use interruptible semaphore acquisition and propagate interruption as an IOException, so it now exposes this race. Both ParallelExecutionTest and IndexBootstrapTest pass.

[JingsongLi]

+1