kaisun2000 commented on code in PR #15373:
URL: https://github.com/apache/druid/pull/15373#discussion_r1431005961
##########
server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java:
##########
@@ -1400,24 +1413,45 @@ public Void apply(@Nullable Object input)
.emit();
}
- droppingSinks.remove(identifier);
- sinkTimeline.remove(
- sink.getInterval(),
- sink.getVersion(),
- identifier.getShardSpec().createChunk(sink)
+ log.info(
+ "Unannounced segment[%s], scheduling drop in [%d] millisecs",
+ identifier,
+ segmentLoaderConfig.getDropSegmentDelayMillis()
);
- for (FireHydrant hydrant : sink) {
- if (cache != null) {
-
cache.close(SinkQuerySegmentWalker.makeHydrantCacheIdentifier(hydrant));
+
+ Runnable removeRunnable = () -> {
+ droppingSinks.remove(identifier);
+ sinkTimeline.remove(
+ sink.getInterval(),
+ sink.getVersion(),
+ identifier.getShardSpec().createChunk(sink)
+ );
+ for (FireHydrant hydrant : sink) {
+ if (cache != null) {
+
cache.close(SinkQuerySegmentWalker.makeHydrantCacheIdentifier(hydrant));
+ }
+ hydrant.swapSegment(null);
}
- hydrant.swapSegment(null);
- }
- if (removeOnDiskData) {
- removeDirectory(computePersistDir(identifier));
- }
+ if (removeOnDiskData) {
+ removeDirectory(computePersistDir(identifier));
+ }
+
+ log.info("Dropped segment[%s].", identifier);
+ };
- log.info("Dropped segment[%s].", identifier);
+ if (segmentLoaderConfig == null) {
+ removeRunnable.run();
+ } else {
+ // Keep the segments in the cache and sinkTimeline for
dropSegmentDelay after unannouncing the segments
+ // This way, in transit queries which still see the segments in
this peon would be able to query the
+ // segments and not throw NullPtr exceptions.
+ exec.schedule(
Review Comment:
How about following the pattern with `persistExecutor`. Basically, in
[shutdownExecutors](https://github.com/apache/druid/blob/master/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java#L1160)
method, I will also call `exec.shutdownNow()`?
--
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]
