src/java/org/apache/cassandra/service/StorageService.java
/**
     * Shuts node off to writes, empties memtables and the commit log.
     * There are two differences between drain and the normal shutdown hook:
     * - Drain waits for in-progress streaming to complete
     * - Drain flushes *all* columnfamilies (shutdown hook only flushes non-durable CFs)
     */
Shuts node off to writes, empties memtables and the commit log.
There are two differences between drain and the normal shutdown hook:
- Drain waits for in-progress streaming to complete
- Drain flushes *all* columnfamilies (shutdown hook only flushes non-durable CFs)
Shuts node off to writes empties memtables and the commit log . There are two differences between drain and the normal shutdown hook : - Drain waits for in - progress streaming to complete - Drain flushes * all * columnfamilies ( shutdown hook only flushes non - durable CFs )
@Override
    public synchronized void drain() throws IOException, InterruptedException, ExecutionException
    {
        ExecutorService mutationStage = StageManager.getStage(Stage.MUTATION);
        if (mutationStage.isTerminated())
        {
            logger_.warn("Cannot drain node (did it already happen?)");
            return;
        }
        setMode(Mode.DRAINING, "starting drain process", true);
        stopRPCServer();
        optionalTasks.shutdown();
        Gossiper.instance.stop();
        setMode(Mode.DRAINING, "shutting down MessageService", false);
        MessagingService.instance().shutdown();
        setMode(Mode.DRAINING, "waiting for streaming", false);
        MessagingService.instance().waitForStreaming();
        setMode(Mode.DRAINING, "clearing mutation stage", false);
        mutationStage.shutdown();
        mutationStage.awaitTermination(3600, TimeUnit.SECONDS);
        StorageProxy.instance.verifyNoHintsInProgress();
        setMode(Mode.DRAINING, "flushing column families", false);
        List<ColumnFamilyStore> cfses = new ArrayList<ColumnFamilyStore>();
        for (String tableName : Schema.instance.getNonSystemTables())
        {
            Table table = Table.open(tableName);
            cfses.addAll(table.getColumnFamilyStores());
        }
        totalCFs = remainingCFs = cfses.size();
        for (ColumnFamilyStore cfs : cfses)
        {
            cfs.forceBlockingFlush();
            remainingCFs--;
        }
        ColumnFamilyStore.postFlushExecutor.shutdown();
        ColumnFamilyStore.postFlushExecutor.awaitTermination(60, TimeUnit.SECONDS);
        CommitLog.instance.shutdownBlocking();
        tasks.shutdown();
        if (!tasks.awaitTermination(1, TimeUnit.MINUTES))
            logger_.warn("Miscellaneous task executor still busy after one minute; proceeding with shutdown");
        setMode(Mode.DRAINED, true);
    }
@ Override public synchronized void drain ( ) throws IOException , InterruptedException , ExecutionException { ExecutorService mutationStage = StageManager . getStage ( Stage . MUTATION ) ; if ( mutationStage . isTerminated ( ) ) { logger_ . warn ( "Cannot drain node (did it already happen?)" ) ; return ; } setMode ( Mode . DRAINING , "starting drain process" , true ) ; stopRPCServer ( ) ; optionalTasks . shutdown ( ) ; Gossiper . instance . stop ( ) ; setMode ( Mode . DRAINING , "shutting down MessageService" , false ) ; MessagingService . instance ( ) . shutdown ( ) ; setMode ( Mode . DRAINING , "waiting for streaming" , false ) ; MessagingService . instance ( ) . waitForStreaming ( ) ; setMode ( Mode . DRAINING , "clearing mutation stage" , false ) ; mutationStage . shutdown ( ) ; mutationStage . awaitTermination ( 3600 , TimeUnit . SECONDS ) ; StorageProxy . instance . verifyNoHintsInProgress ( ) ; setMode ( Mode . DRAINING , "flushing column families" , false ) ; List < ColumnFamilyStore > cfses = new ArrayList < ColumnFamilyStore > ( ) ; for ( String tableName : Schema . instance . getNonSystemTables ( ) ) { Table table = Table . open ( tableName ) ; cfses . addAll ( table . getColumnFamilyStores ( ) ) ; } totalCFs = remainingCFs = cfses . size ( ) ; for ( ColumnFamilyStore cfs : cfses ) { cfs . forceBlockingFlush ( ) ; remainingCFs -- ; } ColumnFamilyStore . postFlushExecutor . shutdown ( ) ; ColumnFamilyStore . postFlushExecutor . awaitTermination ( 60 , TimeUnit . SECONDS ) ; CommitLog . instance . shutdownBlocking ( ) ; tasks . shutdown ( ) ; if ( ! tasks . awaitTermination ( 1 , TimeUnit . MINUTES ) ) logger_ . warn ( "Miscellaneous task executor still busy after one minute; proceeding with shutdown" ) ; setMode ( Mode . DRAINED , true ) ; }
Shuts node off to writes, empties memtables and the commit log.
Shuts node off to writes empties memtables and the commit log .
// wait for miscellaneous tasks like sstable and commitlog segment deletion