private void executeSubBatch ( final int batchId , RebalanceBatchPlanProgressBar progressBar , final Cluster batchRollbackCluster , final List < StoreDefinition > batchRollbackStoreDefs , final List < RebalanceTaskInfo > rebalanceTaskPlanList , boolean hasReadOnlyStores , boolean hasReadWriteStores , boolean finishedReadOnlyStores ) { RebalanceUtils . printBatchLog ( batchId , logger , "Submitting rebalance tasks " ) ; // Get an ExecutorService in place used for submitting our tasks ExecutorService service = RebalanceUtils . createExecutors ( maxParallelRebalancing ) ; // Sub-list of the above list final List < RebalanceTask > failedTasks = Lists . newArrayList ( ) ; final List < RebalanceTask > incompleteTasks = Lists . newArrayList ( ) ; // Semaphores for donor nodes - To avoid multiple disk sweeps Map < Integer , Semaphore > donorPermits = new HashMap < Integer , Semaphore > ( ) ; for ( Node node : batchRollbackCluster . getNodes ( ) ) { donorPermits . put ( node . getId ( ) , new Semaphore ( 1 ) ) ; } try { // List of tasks which will run asynchronously List < RebalanceTask > allTasks = executeTasks ( batchId , progressBar , service , rebalanceTaskPlanList , donorPermits ) ; RebalanceUtils . printBatchLog ( batchId , logger , "All rebalance tasks submitted" ) ; // Wait and shutdown after (infinite) timeout RebalanceUtils . executorShutDown ( service , Long . MAX_VALUE ) ; RebalanceUtils . printBatchLog ( batchId , logger , "Finished waiting for executors" ) ; // Collects all failures + incomplete tasks from the rebalance // tasks. List < Exception > failures = Lists . newArrayList ( ) ; for ( RebalanceTask task : allTasks ) { if ( task . hasException ( ) ) { failedTasks . add ( task ) ; failures . add ( task . getError ( ) ) ; } else if ( ! task . isComplete ( ) ) { incompleteTasks . add ( task ) ; } } if ( failedTasks . size ( ) > 0 ) { throw new VoldemortRebalancingException ( "Rebalance task terminated unsuccessfully on tasks " + failedTasks , failures ) ; } // If there were no failures, then we could have had a genuine // timeout ( Rebalancing took longer than the operator expected ). // We should throw a VoldemortException and not a // VoldemortRebalancingException ( which will start reverting // metadata ). The operator may want to manually then resume the // process. if ( incompleteTasks . size ( ) > 0 ) { throw new VoldemortException ( "Rebalance tasks are still incomplete / running " + incompleteTasks ) ; } } catch ( VoldemortRebalancingException e ) { logger . error ( "Failure while migrating partitions for rebalance task " + batchId ) ; if ( hasReadOnlyStores && hasReadWriteStores && finishedReadOnlyStores ) { // Case 0 adminClient . rebalanceOps . rebalanceStateChange ( null , batchRollbackCluster , null , batchRollbackStoreDefs , null , true , true , false , false , false ) ; } else if ( hasReadWriteStores && finishedReadOnlyStores ) { // Case 4 adminClient . rebalanceOps . rebalanceStateChange ( null , batchRollbackCluster , null , batchRollbackStoreDefs , null , false , true , false , false , false ) ; } throw e ; } finally { if ( ! service . isShutdown ( ) ) { RebalanceUtils . printErrorLog ( batchId , logger , "Could not shutdown service cleanly for rebalance task " + batchId , null ) ; service . shutdownNow ( ) ; } } }
The smallest granularity of rebalancing where - in we move partitions for a sub - set of stores . Finally at the end of the movement the node is removed out of rebalance state