FrankChen021 commented on code in PR #19858:
URL: https://github.com/apache/druid/pull/19858#discussion_r3703837869
##########
extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java:
##########
@@ -204,6 +213,11 @@ public boolean start()
doubleEventCount.incrementAndGet();
LOG.trace("Placed key[%s] val[%s]", key, message);
}
+ if (startingReads.getCount() > 0
Review Comment:
[P2] Wait for the catch-up worker before disposing its cache
When a finite connectTimeout expires while this new catch-up loop is
processing a backlog, start() calls future.cancel(true) and immediately
cacheHandler.close(). Cancellation only interrupts the worker; the record loop
has no interrupt check, so it can continue put/remove calls after close. With
the off-heap cache manager, close deletes the underlying MapDB map, racing
those writes against deletion. Coordinate worker termination, or move cache
disposal into the worker's completion path, before closing the cache.
--
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]
