FrankChen021 commented on code in PR #19858:
URL: https://github.com/apache/druid/pull/19858#discussion_r3712423229
##########
extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java:
##########
@@ -381,6 +387,45 @@ ListenableFuture<?> getFuture()
return future;
}
+ private void shutdownExecutorAndCloseCache()
+ {
+ executorService.shutdown();
+
+ final ListenableFuture<?> future = this.future;
+ if (future != null) {
+ future.cancel(true);
+ }
+
+ boolean interrupted = false;
+ while (!executorService.isTerminated()) {
+ try {
+ executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
Review Comment:
[P1] Do not wait forever after connectTimeout expires
After start() reaches its configured timeout, this call waits indefinitely
for executor termination. future.cancel(true) only requests interruption and
cannot guarantee that consumer code exits; the updated test demonstrates the
regression by configuring a 1 ms timeout yet requiring start() to remain
blocked until an uninterruptible poll is manually released.
LookupReferencesManager.startLookups() waits for every lookup result, so one
stuck Kafka consumer can now block lookup-manager and process startup
indefinitely instead of returning false after connectTimeout. Preserve cache
safety without synchronously waiting forever, for example by transferring cache
disposal to worker completion while allowing start() to return after bounded
cancellation.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
