protected void abandonSegment ( final long truncatedTime , final Sink sink ) { if ( sinks . containsKey ( truncatedTime ) ) { try { segmentAnnouncer . unannounceSegment ( sink . getSegment ( ) ) ; removeSegment ( sink , computePersistDir ( schema , sink . getInterval ( ) ) ) ; log . info ( "Removing sinkKey %d for segment %s" , truncatedTime , sink . getSegment ( ) . getId ( ) ) ; sinks . remove ( truncatedTime ) ; metrics . setSinkCount ( sinks . size ( ) ) ; sinkTimeline . remove ( sink . getInterval ( ) , sink . getVersion ( ) , new SingleElementPartitionChunk <> ( sink ) ) ; for ( FireHydrant hydrant : sink ) { cache . close ( SinkQuerySegmentWalker . makeHydrantCacheIdentifier ( hydrant ) ) ; hydrant . swapSegment ( null ) ; } synchronized ( handoffCondition ) { handoffCondition . notifyAll ( ) ; } } catch ( Exception e ) { log . makeAlert ( e , "Unable to abandon old segment for dataSource[%s]" , schema . getDataSource ( ) ) . addData ( "interval" , sink . getInterval ( ) ) . emit ( ) ; } } }
Unannounces a given sink and removes all local references to it . It is important that this is only called from the single - threaded mergeExecutor since otherwise chaos may ensue if merged segments are deleted while being created .