apoorvmittal10 commented on code in PR #21964:
URL: https://github.com/apache/kafka/pull/21964#discussion_r4115539955
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java:
##########
@@ -127,20 +127,25 @@ public void setup() throws IOException,
RemoteStorageException {
}
@AfterEach
- public void cleanup() throws InterruptedException {
+ public void cleanup() throws Exception {
reset(rsm);
- // the files created for the test will be deleted automatically on
thread exit since we use temp dir
- Utils.closeQuietly(cache, "RemoteIndexCache created for unit test");
- // best effort to delete the per-test resource. Even if we don't
delete, it is ok because the parent directory
- // will be deleted at the end of test.
+ // Close the cache directly instead of using closeQuietly to ensure
any exception during close
+ // (e.g. failure to shut down the cleaner scheduler) is properly
propagated rather than silently swallowed,
+ // which could leave the cleaner thread alive and cause the leaked
thread assertion below to fail.
try {
- Utils.delete(logDir);
- } catch (IOException ioe) {
- // ignore
+ cache.close();
+ } finally {
+ // best effort to delete the per-test resource. Even if we don't
delete, it is ok because the parent directory
+ // will be deleted at the end of test.
+ try {
+ Utils.delete(logDir);
+ } catch (IOException ioe) {
+ // ignore
+ }
+ // Verify no lingering threads. It is important to have this as
the very last statement in the @AfterEach
+ // because this may throw an exception and prevent cleanup after it
+
TestUtils.assertNoLeakedThreadsWithNameAndDaemonStatus(REMOTE_LOG_INDEX_CACHE_CLEANER_THREAD,
true);
Review Comment:
Any reason of moving this inside the try block now? Can live outside a
earlier, right?
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java:
##########
@@ -1044,6 +1049,11 @@ public void testConcurrentCacheDeletedFileExists()
throws InterruptedException,
"Time index file should not be present on disk at " +
remoteIndexCacheDir.toPath());
TestUtils.waitForCondition(() -> getIndexFileFromRemoteCacheDir(cache,
LogFileUtils.DELETED_FILE_SUFFIX).isEmpty(),
"Index file marked for deletion should not be present on disk
at " + remoteIndexCacheDir.toPath());
+
+ // Wait for all scheduled cleanup tasks to complete so the cleaner
thread is idle before @AfterEach
+ // shuts it down. This prevents a race between an in-flight task and
scheduler shutdown.
+ TestUtils.waitForCondition(() -> cache.expiredIdxPendingForDeletion()
== 0,
Review Comment:
It checks the scheduler's queue size (pendingTaskSize()), which hits 0 as
soon as a worker dequeues the task but does it guarantee that cleanup actually
finished?
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java:
##########
@@ -127,20 +127,25 @@ public void setup() throws IOException,
RemoteStorageException {
}
@AfterEach
- public void cleanup() throws InterruptedException {
+ public void cleanup() throws Exception {
reset(rsm);
- // the files created for the test will be deleted automatically on
thread exit since we use temp dir
- Utils.closeQuietly(cache, "RemoteIndexCache created for unit test");
- // best effort to delete the per-test resource. Even if we don't
delete, it is ok because the parent directory
- // will be deleted at the end of test.
+ // Close the cache directly instead of using closeQuietly to ensure
any exception during close
+ // (e.g. failure to shut down the cleaner scheduler) is properly
propagated rather than silently swallowed,
+ // which could leave the cleaner thread alive and cause the leaked
thread assertion below to fail.
try {
- Utils.delete(logDir);
- } catch (IOException ioe) {
- // ignore
+ cache.close();
+ } finally {
+ // best effort to delete the per-test resource. Even if we don't
delete, it is ok because the parent directory
+ // will be deleted at the end of test.
+ try {
+ Utils.delete(logDir);
+ } catch (IOException ioe) {
+ // ignore
+ }
+ // Verify no lingering threads. It is important to have this as
the very last statement in the @AfterEach
+ // because this may throw an exception and prevent cleanup after it
+
TestUtils.assertNoLeakedThreadsWithNameAndDaemonStatus(REMOTE_LOG_INDEX_CACHE_CLEANER_THREAD,
true);
Review Comment:
If cache.close throws then still there can be a leak, correct?
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java:
##########
@@ -127,20 +127,25 @@ public void setup() throws IOException,
RemoteStorageException {
}
@AfterEach
- public void cleanup() throws InterruptedException {
+ public void cleanup() throws Exception {
reset(rsm);
- // the files created for the test will be deleted automatically on
thread exit since we use temp dir
- Utils.closeQuietly(cache, "RemoteIndexCache created for unit test");
- // best effort to delete the per-test resource. Even if we don't
delete, it is ok because the parent directory
- // will be deleted at the end of test.
+ // Close the cache directly instead of using closeQuietly to ensure
any exception during close
+ // (e.g. failure to shut down the cleaner scheduler) is properly
propagated rather than silently swallowed,
+ // which could leave the cleaner thread alive and cause the leaked
thread assertion below to fail.
try {
- Utils.delete(logDir);
- } catch (IOException ioe) {
- // ignore
+ cache.close();
Review Comment:
Do we require null check for cache? If setup i.e. before each fails before
cache is instantiated then can there be a NPE here?
--
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]
