server/src/main/java/org/apache/druid/server/coordination/SegmentLoadDropHandler.java
/**
   * Bulk adding segments during bootstrap
   * @param segments A collection of segments to add
   * @param callback Segment loading callback
   */
Bulk adding segments during bootstrap
@param segments A collection of segments to add
@param callback Segment loading callback
Bulk adding segments during bootstrap @param segments A collection of segments to add @param callback Segment loading callback
private void addSegments(Collection<DataSegment> segments, final DataSegmentChangeCallback callback)
  {
    ExecutorService loadingExecutor = null;
    ExecutorService loadSegmentsIntoPageCacheOnBootstrapExec =
        config.getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap() != 0 ?
        Execs.multiThreaded(config.getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap(),
                            "Load-Segments-Into-Page-Cache-On-Bootstrap-%s") : null;
    try (final BackgroundSegmentAnnouncer backgroundSegmentAnnouncer =
             new BackgroundSegmentAnnouncer(announcer, exec, config.getAnnounceIntervalMillis())) {
      backgroundSegmentAnnouncer.startAnnouncing();
      loadingExecutor = Execs.multiThreaded(config.getNumBootstrapThreads(), "Segment-Load-Startup-%s");
      final int numSegments = segments.size();
      final CountDownLatch latch = new CountDownLatch(numSegments);
      final AtomicInteger counter = new AtomicInteger(0);
      final CopyOnWriteArrayList<DataSegment> failedSegments = new CopyOnWriteArrayList<>();
      for (final DataSegment segment : segments) {
        loadingExecutor.submit(
            () -> {
              try {
                log.info(
                    "Loading segment[%d/%d][%s]",
                    counter.incrementAndGet(),
                    numSegments,
                    segment.getId()
                );
                loadSegment(segment, callback, config.isLazyLoadOnStart(), loadSegmentsIntoPageCacheOnBootstrapExec);
                try {
                  backgroundSegmentAnnouncer.announceSegment(segment);
                }
                catch (InterruptedException e) {
                  Thread.currentThread().interrupt();
                  throw new SegmentLoadingException(e, "Loading Interrupted");
                }
              }
              catch (SegmentLoadingException e) {
                log.error(e, "[%s] failed to load", segment.getId());
                failedSegments.add(segment);
              }
              finally {
                latch.countDown();
              }
            }
        );
      }
      try {
        latch.await();
        if (failedSegments.size() > 0) {
          log.makeAlert("%,d errors seen while loading segments", failedSegments.size())
             .addData("failedSegments", failedSegments)
             .emit();
        }
      }
      catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        log.makeAlert(e, "LoadingInterrupted").emit();
      }
      backgroundSegmentAnnouncer.finishAnnouncing();
    }
    catch (SegmentLoadingException e) {
      log.makeAlert(e, "Failed to load segments -- likely problem with announcing.")
         .addData("numSegments", segments.size())
         .emit();
    }
    finally {
      callback.execute();
      if (loadingExecutor != null) {
        loadingExecutor.shutdownNow();
      }
      if (loadSegmentsIntoPageCacheOnBootstrapExec != null) {
        loadSegmentsIntoPageCacheOnBootstrapExec.shutdown();
      }
    }
  }
private void addSegments ( Collection < DataSegment > segments , final DataSegmentChangeCallback callback ) { ExecutorService loadingExecutor = null ; ExecutorService loadSegmentsIntoPageCacheOnBootstrapExec = config . getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap ( ) != 0 ? Execs . multiThreaded ( config . getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap ( ) , "Load-Segments-Into-Page-Cache-On-Bootstrap-%s" ) : null ; try ( final BackgroundSegmentAnnouncer backgroundSegmentAnnouncer = new BackgroundSegmentAnnouncer ( announcer , exec , config . getAnnounceIntervalMillis ( ) ) ) { backgroundSegmentAnnouncer . startAnnouncing ( ) ; loadingExecutor = Execs . multiThreaded ( config . getNumBootstrapThreads ( ) , "Segment-Load-Startup-%s" ) ; final int numSegments = segments . size ( ) ; final CountDownLatch latch = new CountDownLatch ( numSegments ) ; final AtomicInteger counter = new AtomicInteger ( 0 ) ; final CopyOnWriteArrayList < DataSegment > failedSegments = new CopyOnWriteArrayList < > ( ) ; for ( final DataSegment segment : segments ) { loadingExecutor . submit ( ( ) -> { try { log . info ( "Loading segment[%d/%d][%s]" , counter . incrementAndGet ( ) , numSegments , segment . getId ( ) ) ; loadSegment ( segment , callback , config . isLazyLoadOnStart ( ) , loadSegmentsIntoPageCacheOnBootstrapExec ) ; try { backgroundSegmentAnnouncer . announceSegment ( segment ) ; } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; throw new SegmentLoadingException ( e , "Loading Interrupted" ) ; } } catch ( SegmentLoadingException e ) { log . error ( e , "[%s] failed to load" , segment . getId ( ) ) ; failedSegments . add ( segment ) ; } finally { latch . countDown ( ) ; } } ) ; } try { latch . await ( ) ; if ( failedSegments . size ( ) > 0 ) { log . makeAlert ( "%,d errors seen while loading segments" , failedSegments . size ( ) ) . addData ( "failedSegments" , failedSegments ) . emit ( ) ; } } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; log . makeAlert ( e , "LoadingInterrupted" ) . emit ( ) ; } backgroundSegmentAnnouncer . finishAnnouncing ( ) ; } catch ( SegmentLoadingException e ) { log . makeAlert ( e , "Failed to load segments -- likely problem with announcing." ) . addData ( "numSegments" , segments . size ( ) ) . emit ( ) ; } finally { callback . execute ( ) ; if ( loadingExecutor != null ) { loadingExecutor . shutdownNow ( ) ; } if ( loadSegmentsIntoPageCacheOnBootstrapExec != null ) { loadSegmentsIntoPageCacheOnBootstrapExec . shutdown ( ) ; } } }
Bulk adding segments during bootstrap
 @param segments A collection of segments to add
 @param callback Segment loading callback
Bulk adding segments during bootstrap @param segments A collection of segments to add @param callback Segment loading callback
// Start a temporary thread pool to load segments into page cache during bootstrap // At this stage, all tasks have been submitted, send a shutdown command to the bootstrap // thread pool so threads will exit after finishing the tasks