start or stop MigratePartitionLeader task
489473
void processMigratePartitionLeaderTask(MigratePartitionLeaderMessage message) {
        synchronized(m_lock) {
            //start MigratePartitionLeader service
            if (message.startMigratingPartitionLeaders()) {
                if (m_migratePartitionLeaderExecutor == null) {
                    m_migratePartitionLeaderExecutor = Executors.newSingleThreadScheduledExecutor(CoreUtils.getThreadFactory("MigratePartitionLeader"));
                    final int interval = Integer.parseInt(System.getProperty("MIGRATE_PARTITION_LEADER_INTERVAL", "1"));
                    final int delay = Integer.parseInt(System.getProperty("MIGRATE_PARTITION_LEADER_DELAY", "1"));
                    m_migratePartitionLeaderExecutor.scheduleAtFixedRate(
                            () -> startMigratePartitionLeader(message.isForStopNode()),
                            delay, interval, TimeUnit.SECONDS);
                }
                hostLog.info("MigratePartitionLeader task is started.");
                return;
            }

            //stop MigratePartitionLeader service
            if (m_migratePartitionLeaderExecutor != null ) {
                m_migratePartitionLeaderExecutor.shutdown();
                m_migratePartitionLeaderExecutor = null;
            }
        }
        hostLog.info("MigratePartitionLeader task is stopped.");
    }
530370
@Override
  public SerializableVoid runTask(MigrateConfig config, ArrayList<MigrateCommand> commands,
      RunTaskContext context) throws Exception {
    WriteType writeType = config.getWriteType() == null
        ? ServerConfiguration.getEnum(PropertyKey.USER_FILE_WRITE_TYPE_DEFAULT, WriteType.class)
        : WriteType.valueOf(config.getWriteType());
    for (MigrateCommand command : commands) {
      migrate(command, writeType.toProto(), config.isDeleteSource(),
          context.getFileSystem());
    }
    // Try to delete the source directory if it is empty.
    if (config.isDeleteSource() && !hasFiles(new AlluxioURI(config.getSource()),
        context.getFileSystem())) {
      try {
        LOG.debug("Deleting {}", config.getSource());
        context.getFileSystem().delete(new AlluxioURI(config.getSource()),
            DeletePOptions.newBuilder().setRecursive(true).build());
      } catch (FileDoesNotExistException e) {
        // It's already deleted, possibly by another worker.
      }
    }
    return null;
  }
810570
private void triggerMasterToAssignPartitions() {
        if (!shouldTriggerMasterToAssignPartitions()) {
            return;
        }

        ClusterServiceImpl clusterService = node.getClusterService();

        ClusterState clusterState = clusterService.getClusterState();
        if (!clusterState.isMigrationAllowed()) {
            logger.warning("Partitions can't be assigned since cluster-state=" + clusterState);
            return;
        }

        final Address masterAddress = clusterService.getMasterAddress();
        if (masterAddress == null || masterAddress.equals(node.getThisAddress())) {
            return;
        }

        if (masterTriggered.compareAndSet(false, true)) {
            InternalOperationService operationService = nodeEngine.getOperationService();
            ICompletableFuture<PartitionRuntimeState> future =
                    operationService.invokeOnTarget(SERVICE_NAME, new AssignPartitions(), masterAddress);
            future.andThen(new ExecutionCallback<PartitionRuntimeState>() {
                @Override
                public void onResponse(PartitionRuntimeState partitionState) {
                    resetMasterTriggeredFlag();
                    if (partitionState != null) {
                        partitionState.setMaster(masterAddress);
                        processPartitionRuntimeState(partitionState);
                    }
                }

                @Override
                public void onFailure(Throwable t) {
                    resetMasterTriggeredFlag();
                    logger.severe(t);
                }
            });

            masterTrigger.executeWithDelay();
        }
    }
1362297
public void startPartitionWatch() throws InterruptedException, ExecutionException {
        Future<?> task = m_es.submit(new PartitionWatchEvent());
        task.get();
    }
639263
public static MigratePartitionLeaderInfo getMigratePartitionLeaderInfo(ZooKeeper zk) {
        try {
            byte[] data = zk.getData(migrate_partition_leader_info, null, null);
            if (data != null) {
                MigratePartitionLeaderInfo info = new MigratePartitionLeaderInfo(data);
                return info;
            }
        } catch (KeeperException | InterruptedException | JSONException e) {
        }
        return null;
    }
317662
private void moveToNextPartition() {
    if (this.currentPartitionIdx == INITIAL_PARTITION_IDX) {
      LOG.info("Pulling topic " + this.topicName);
      this.currentPartitionIdx = 0;
    } else {
      updateStatisticsForCurrentPartition();
      this.currentPartitionIdx++;
      this.currentPartitionRecordCount = 0;
      this.currentPartitionTotalSize = 0;
      this.currentPartitionDecodeRecordTime = 0;
      this.currentPartitionFetchMessageBufferTime = 0;
      this.currentPartitionReadRecordTime = 0;
      this.currentPartitionLastSuccessfulRecord = null;
    }

    this.messageIterator = null;
    if (this.currentPartitionIdx < this.partitions.size()) {
      LOG.info(String.format("Pulling partition %s from offset %d to %d, range=%d", this.getCurrentPartition(),
          this.nextWatermark.get(this.currentPartitionIdx), this.highWatermark.get(this.currentPartitionIdx),
          this.highWatermark.get(this.currentPartitionIdx) - this.nextWatermark.get(this.currentPartitionIdx)));
      switchMetricContextToCurrentPartition();
    }

    if (!allPartitionsFinished()) {
      this.startFetchEpochTime.put(this.getCurrentPartition(), System.currentTimeMillis());
    }
  }
1106757
def run_command(self, cluster_topology, cluster_balancer):
        """"""

        # The ideal weight of each broker is total_weight / broker_count.
        # It should be possible to remove partitions from each broker until
        # the weight of the broker is less than this ideal value, otherwise it
        # is impossible to balance the cluster. If --max-movement-size is too
        # small, exit with an error.
        if self.args.max_movement_size:
            total_weight = sum(
                partition.weight
                for partition in six.itervalues(cluster_topology.partitions)
            )
            broker_count = len(cluster_topology.brokers)
            optimal_weight = total_weight / broker_count

            broker, max_unmovable_on_one_broker = max((
                (broker, sum(
                    partition.weight
                    for partition in broker.partitions
                    if partition.size > self.args.max_movement_size
                ))
                for broker in cluster_topology.brokers.values()),
                key=lambda t: t[1],
            )

            if max_unmovable_on_one_broker >= optimal_weight:
                sorted_partitions = sorted(
                    [
                        partition
                        for partition in broker.partitions
                        if partition.size > self.args.max_movement_size
                    ],
                    reverse=True,
                    key=lambda partition: partition.size,
                )

                for partition in sorted_partitions:
                    max_unmovable_on_one_broker -= partition.weight
                    if max_unmovable_on_one_broker <= optimal_weight:
                        required_max_movement_size = partition.size
                        break

                self.log.error(
                    'Max movement size {max_movement_size} is too small, it is'
                    ' not be possible to balance the cluster. A max movement'
                    ' size of {required} or higher is required.'.format(
                        max_movement_size=self.args.max_movement_size,
                        required=required_max_movement_size,
                    )
                )
                sys.exit(1)
        elif self.args.auto_max_movement_size:
            self.args.max_movement_size = max(
                partition.size
                for partition in six.itervalues(cluster_topology.partitions)
            )
            self.log.info(
                'Auto-max-movement-size: using {max_movement_size} as'
                ' max-movement-size.'.format(
                    max_movement_size=self.args.max_movement_size,
                )
            )

        base_assignment = cluster_topology.assignment
        base_score = cluster_balancer.score()
        rg_imbalance, _ = get_replication_group_imbalance_stats(
            list(cluster_topology.rgs.values()),
            list(cluster_topology.partitions.values())
        )

        cluster_balancer.rebalance()

        assignment = cluster_topology.assignment
        score = cluster_balancer.score()
        new_rg_imbalance, _ = get_replication_group_imbalance_stats(
            list(cluster_topology.rgs.values()),
            list(cluster_topology.partitions.values())
        )

        if self.args.show_stats:
            display_cluster_topology_stats(cluster_topology, base_assignment)
            if base_score is not None and score is not None:
                print('\nScore before: %f' % base_score)
                print('Score after:  %f' % score)
                print('Score improvement: %f' % (score - base_score))

        if not validate_plan(
            assignment_to_plan(assignment),
            assignment_to_plan(base_assignment),
        ):
            self.log.error('Invalid latest-cluster assignment. Exiting.')
            sys.exit(1)

        if self.args.score_improvement_threshold:
            if base_score is None or score is None:
                self.log.error(
                    '%s cannot assign scores so --score-improvement-threshold'
                    ' cannot be used.',
                    cluster_balancer.__class__.__name__,
                )
                return
            else:
                score_improvement = score - base_score
                if score_improvement >= self.args.score_improvement_threshold:
                    self.log.info(
                        'Score improvement %f is greater than the threshold %f.'
                        ' Continuing to apply the assignment.',
                        score_improvement,
                        self.args.score_improvement_threshold,
                    )
                elif new_rg_imbalance < rg_imbalance:
                    self.log.info(
                        'Score improvement %f is less than the threshold %f,'
                        ' but replica balance has improved. Continuing to'
                        ' apply the assignment.',
                        score_improvement,
                        self.args.score_improvement_threshold,
                    )
                else:
                    self.log.info(
                        'Score improvement %f is less than the threshold %f.'
                        ' Assignment will not be applied.',
                        score_improvement,
                        self.args.score_improvement_threshold,
                    )
                    return

        # Reduce the proposed assignment based on max_partition_movements
        # and max_leader_changes
        reduced_assignment = self.get_reduced_assignment(
            base_assignment,
            cluster_topology,
            self.args.max_partition_movements,
            self.args.max_leader_changes,
        )
        if reduced_assignment:
            self.process_assignment(reduced_assignment)
        else:
            self.log.info("Cluster already balanced. No actions to perform.")
821843
void sendMigrationEvent(final MigrationInfo migrationInfo, final MigrationEvent.MigrationStatus status) {
        if (migrationInfo.getSourceCurrentReplicaIndex() != 0
                && migrationInfo.getDestinationNewReplicaIndex() != 0) {
            // only fire events for 0th replica migrations
            return;
        }

        ClusterServiceImpl clusterService = node.getClusterService();
        MemberImpl current = clusterService.getMember(migrationInfo.getSourceAddress());
        MemberImpl newOwner = clusterService.getMember(migrationInfo.getDestinationAddress());
        MigrationEvent event = new MigrationEvent(migrationInfo.getPartitionId(), current, newOwner, status);
        EventService eventService = nodeEngine.getEventService();
        Collection<EventRegistration> registrations = eventService.getRegistrations(SERVICE_NAME, MIGRATION_EVENT_TOPIC);
        eventService.publishEvent(SERVICE_NAME, registrations, event, event.getPartitionId());
    }
708285
private void doRun() {
        if (migrationInfo.startProcessing()) {
            try {
                if (firstFragment) {
                    executeBeforeMigrations();
                }

                for (Operation migrationOperation : fragmentMigrationState.getMigrationOperations()) {
                    runMigrationOperation(migrationOperation);
                }

                success = true;
            } catch (Throwable e) {
                failureReason = e;
                getLogger().severe("Error while executing replication operations " + migrationInfo, e);
            } finally {
                afterMigrate();
            }
        } else {
            logMigrationCancelled();
        }
    }
1087260
@Override
    public Future<?> startPartition(PartitionPlanConfig partitionPlanConfig,
                                    Step step,
                                    PartitionReplyQueue partitionReplyQueue) {
        //This partition was dispatched remotely. Hence, param isRemoteDispatch should be true
        Entry<BatchPartitionWorkUnit, Future<?>> workUnitFutureEntry = batchKernelService.startPartition(partitionPlanConfig, step, partitionReplyQueue, true);
        return workUnitFutureEntry.getValue();
    }
519846
public void stop() throws Exception {
		LOG.info("Stop job leader service.");

		if (JobLeaderService.State.STARTED == state) {

			for (Tuple2<LeaderRetrievalService, JobLeaderService.JobManagerLeaderListener> leaderRetrievalServiceEntry: jobLeaderServices.values()) {
				LeaderRetrievalService leaderRetrievalService = leaderRetrievalServiceEntry.f0;
				JobLeaderService.JobManagerLeaderListener jobManagerLeaderListener = leaderRetrievalServiceEntry.f1;

				jobManagerLeaderListener.stop();
				leaderRetrievalService.stop();
			}

			jobLeaderServices.clear();
		}

		state = JobLeaderService.State.STOPPED;
	}
277017
function() {
            if (_.size(this.tasks) > 0) {
                this.empty = false;
                var task = this.tasks.shift();
                this.startTask(task);
                this.trigger("tasks:next");
            } else {
                this.empty = true;
                this.trigger("tasks:finish");
            }      
        }
1129350
def _start(self, tpart):
        """"""
        g = gevent.Greenlet(self.uploader, tpart)
        g.link(self._finish)

        # Account for concurrency_burden before starting the greenlet
        # to avoid racing against .join.
        self.concurrency_burden += 1

        self.member_burden += len(tpart)

        g.start()
1156124
private void moveTaskOutputs(TaskAttemptContext context,
                               FileSystem fs,
                               Path jobOutputDir,
                               Path taskOutput) 
  throws IOException {
    TaskAttemptID attemptId = context.getTaskAttemptID();
    context.progress();
    if (fs.isFile(taskOutput)) {
      Path finalOutputPath = getFinalPath(jobOutputDir, taskOutput, 
                                          workPath);
      if (!fs.rename(taskOutput, finalOutputPath)) {
        if (!fs.delete(finalOutputPath, true)) {
          throw new IOException("Failed to delete earlier output of task: " + 
                                 attemptId);
        }
        if (!fs.rename(taskOutput, finalOutputPath)) {
          throw new IOException("Failed to save output of task: " + 
        		  attemptId);
        }
      }
      LOG.debug("Moved " + taskOutput + " to " + finalOutputPath);
    } else if(fs.getFileStatus(taskOutput).isDir()) {
      FileStatus[] paths = fs.listStatus(taskOutput);
      Path finalOutputPath = getFinalPath(jobOutputDir, taskOutput, workPath);
      fs.mkdirs(finalOutputPath);
      if (paths != null) {
        for (FileStatus path : paths) {
          moveTaskOutputs(context, fs, jobOutputDir, path.getPath());
        }
      }
    }
  }
981761
protected void taskStart(int taskId, int rebalanceAsyncId) {
        taskCompletionTimeMs = System.currentTimeMillis();
        taskLog("[TaskId : " + taskId + " ] Starting rebalance of " + partitionStoreCount
                + " partition-stores for async operation id " + rebalanceAsyncId + ".");
        progressBar.beginTask(taskId);
    }
234754
public static boolean createMigratePartitionLeaderInfo(ZooKeeper zk, MigratePartitionLeaderInfo info) {
        try {
            zk.create(migrate_partition_leader_info, info.toBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (KeeperException e) {
            if (e.code() == KeeperException.Code.NODEEXISTS) {
                try {
                    zk.setData(migrate_partition_leader_info, info.toBytes(), -1);
                } catch (KeeperException | InterruptedException | JSONException e1) {
                }
                return false;
            }

            org.voltdb.VoltDB.crashLocalVoltDB("Unable to create MigratePartitionLeader Indicator", true, e);
        } catch (InterruptedException | JSONException e) {
            org.voltdb.VoltDB.crashLocalVoltDB("Unable to create MigratePartitionLeader Indicator", true, e);
        }

        return true;
    }
341581
@Override
  void enableTaskForCleanup(PathDeletionContext context)
      throws IOException {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Going to do " + TaskCommands.ENABLE_TASK_FOR_CLEANUP.toString()
                + " for " + context.fullPath);
    }

    if (context instanceof TaskControllerPathDeletionContext) {
      TaskControllerPathDeletionContext tContext =
        (TaskControllerPathDeletionContext) context;
    
      if (tContext.task.getUser() != null &&
          tContext.fs instanceof LocalFileSystem) {
        try {
          runCommand(TaskCommands.ENABLE_TASK_FOR_CLEANUP,
                   tContext.task.getUser(),
                   buildTaskCleanupArgs(tContext), null, null);
        } catch(IOException e) {
          LOG.warn("Uanble to change permissions for " + tContext.fullPath);
        }
      }
      else {
        throw new IllegalArgumentException("Either user is null or the "  +
                               "file system is not local file system.");
      }
    }
    else {
      throw new IllegalArgumentException("PathDeletionContext provided is not "
          + "TaskControllerPathDeletionContext.");
    }
  }