divijvaidya commented on code in PR #13535: URL: https://github.com/apache/kafka/pull/13535#discussion_r1182398774
########## core/src/main/java/kafka/log/remote/RemoteLogManager.java: ########## @@ -670,6 +875,14 @@ public void close() { } catch (InterruptedException e) { // ignore } + remoteStorageReaderThreadPool.shutdownNow(); + //waits for 2 mins to terminate the current tasks + try { + remoteStorageReaderThreadPool.awaitTermination(2, TimeUnit.MINUTES); Review Comment: How did we decide on 2min. here? I don't think we should block shutdown of the broker on this over here because there are other limits associated with clean vs unclean shutdown. If we do plan to block, we should tie it to overall shutdown timeout. As an example, clean shutdown is expected to be completed in 5 min. see `lifecycleManager.controlledShutdownFuture.get(5L, TimeUnit.MINUTES)` in BrokerServer.scala. ########## core/src/main/java/kafka/log/remote/RemoteLogManager.java: ########## @@ -670,6 +875,14 @@ public void close() { } catch (InterruptedException e) { // ignore } + remoteStorageReaderThreadPool.shutdownNow(); Review Comment: 1. Should we try to complete the pending reads first using `shutdown()` instead of interrupting them via `shutdownNow()`? This would potentially reduce abrupt behaviour if RSM doesn't handle interruption correctly. I will suggest the following pattern (also see `shutdownAndAwaitTermination` at https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ExecutorService.html) ``` remoteStorageReaderThreadPool.shutdown(); // stop accepting new requests to this pool try { // wait until the pool has terminated if(!remoteStorageReaderThreadPool.awaitTermination(2, TimeUnit.MINUTES)) { LOGGER.warn("Shutdown for remoteStorageReaderThreadPool didn't complete all tasks."); remoteStorageReaderThreadPool.shutdownNow(); } } catch (InterruptedException e) { remoteStorageReaderThreadPool.shutdownNow(); LOGGER.warn("Shutdown for remoteStorageReaderThreadPool interrupted"); } LOGGER.info("RemoteStorageReaderThreadPool stopped."); ``` 2. I would also suggest to explain why aren't we re-interrupting the current thread using Thread.currentThread().interrupt(). 3. Please add warn and info logs that help debug the correct order of shutdown. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org