Called on the master node to publish the current partition state to all cluster nodes. It will not publish the partition
state if the partitions have not yet been initialized, there is ongoing repartitioning or a node is joining the cluster.
517611
@SuppressWarnings("checkstyle:npathcomplexity")
    void publishPartitionRuntimeState() {
        if (!partitionStateManager.isInitialized()) {
            // do not send partition state until initialized!
            return;
        }

        if (!node.isMaster()) {
            return;
        }

        if (!areMigrationTasksAllowed()) {
            // migration is disabled because of a member leave, wait till enabled!
            return;
        }

        PartitionRuntimeState partitionState = createPartitionStateInternal();
        if (partitionState == null) {
            return;
        }

        if (logger.isFineEnabled()) {
            logger.fine("Publishing partition state, version: " + partitionState.getVersion());
        }

        PartitionStateOperation op = new PartitionStateOperation(partitionState, false);
        OperationService operationService = nodeEngine.getOperationService();
        Collection<Member> members = node.clusterService.getMembers();
        for (Member member : members) {
            if (!member.localMember()) {
                try {
                    operationService.send(op, member.getAddress());
                } catch (Exception e) {
                    logger.finest(e);
                }
            }
        }
    }
793053
@Override
	public void publish(Event event) {

		// Check state
		if (status.get() == STATUS_STOPPED) {
			return;
		}

		// Connected
		if (event instanceof ConnectedEvent) {
			if (status.compareAndSet(STATUS_CONNECTING, STATUS_CONNECTED)) {

				// Redis connection is Ok
				logger.info("Redis get-set connection are estabilished.");
			}
			return;
		}

		// Disconnected
		if (event instanceof DisconnectedEvent) {
			int s = status.getAndSet(STATUS_DISCONNECTED);
			if (s != STATUS_DISCONNECTED) {
				logger.info("Redis get-set connection aborted.");
				reconnect();
			}
		}

	}
1204867
def _wait_nodes_updated(self, cluster_member, new_nodes):
        '''
        '''
        while True:
            known_nodes = [n['name'] for n in cluster_member.nodes()]
            added_nodes = [n.name for n in new_nodes]
            if len(set(known_nodes) & set(added_nodes)) == len(added_nodes):
                break;
            logger.info('waiting for adding new nodes in `cluster nodes`')
            time.sleep(1)
640742
void publishTaskState(String taskName, Map stateSnapshot, ScheduledTaskStatisticsImpl statsSnapshot,
                          ScheduledTaskResult result) {
        if (logger.isFinestEnabled()) {
            log(FINEST, "Publishing state, to replicas. State: " + stateSnapshot);
        }

        Operation op = new SyncStateOperation(getName(), taskName, stateSnapshot, statsSnapshot, result);
        createInvocationBuilder(op)
                .invoke()
                .join();
    }
728806
private synchronized void onStateChange(PartitionId partitionId, PrimitiveState state) {
    states.put(partitionId, state);
    switch (state) {
      case CONNECTED:
        if (this.state != PrimitiveState.CONNECTED && !states.containsValue(PrimitiveState.SUSPENDED) && !states.containsValue(PrimitiveState.CLOSED)) {
          this.state = PrimitiveState.CONNECTED;
          stateChangeListeners.forEach(l -> l.accept(PrimitiveState.CONNECTED));
        }
        break;
      case SUSPENDED:
        if (this.state == PrimitiveState.CONNECTED) {
          this.state = PrimitiveState.SUSPENDED;
          stateChangeListeners.forEach(l -> l.accept(PrimitiveState.SUSPENDED));
        }
        break;
      case CLOSED:
        if (this.state != PrimitiveState.CLOSED) {
          this.state = PrimitiveState.CLOSED;
          stateChangeListeners.forEach(l -> l.accept(PrimitiveState.CLOSED));
        }
        break;
    }
  }
81641
private Pair<Cluster, List<StoreDefinition>> getCurrentClusterState() {

        // Retrieve the latest cluster metadata from the existing nodes
        Versioned<Cluster> currentVersionedCluster = adminClient.rebalanceOps.getLatestCluster(Utils.nodeListToNodeIdList(Lists.newArrayList(adminClient.getAdminClientCluster()
                                                                                                                                                        .getNodes())));
        Cluster cluster = currentVersionedCluster.getValue();
        List<StoreDefinition> storeDefs = adminClient.rebalanceOps.getCurrentStoreDefinitions(cluster);
        return new Pair<Cluster, List<StoreDefinition>>(cluster, storeDefs);
    }
408906
private boolean preCheckClusterState(ClusterService clusterService) {
        ClusterState initialState = clusterService.getClusterState();
        if (!initialState.isJoinAllowed()) {
            logger.warning("Could not prepare cluster state since it has been changed to " + initialState);
            return false;
        }

        return true;
    }
959943
public void updatePartitionLeader(int partitionId, long newMasterHISD, boolean isLeaderMigrated) {
        PartitionCallback cb = m_callbacks.get(partitionId);
        if (cb != null && cb.m_currentLeader != newMasterHISD) {
            cb.m_previousLeader = cb.m_currentLeader;
            cb.m_currentLeader = newMasterHISD;
            cb.m_isLeaderMigrated = isLeaderMigrated;
        }
    }
708464
public boolean processPartitionRuntimeState(final PartitionRuntimeState partitionState) {
        Address sender = partitionState.getMaster();
        if (!node.getNodeExtension().isStartCompleted()) {
            logger.warning("Ignoring received partition table, startup is not completed yet. Sender: " + sender);
            return false;
        }

        if (!validateSenderIsMaster(sender, "partition table update")) {
            return false;
        }

        return applyNewPartitionTable(partitionState.getPartitionTable(), partitionState.getVersion(),
                partitionState.getCompletedMigrations(), sender);
    }
1201507
def commit(self, partitions=None):
        

        # short circuit if nothing happened. This check is kept outside
        # to prevent un-necessarily acquiring a lock for checking the state
        if self.count_since_commit == 0:
            return

        with self.commit_lock:
            # Do this check again, just in case the state has changed
            # during the lock acquiring timeout
            if self.count_since_commit == 0:
                return

            reqs = []
            if partitions is None:  # commit all partitions
                partitions = list(self.offsets.keys())

            log.debug('Committing new offsets for %s, partitions %s',
                      self.topic, partitions)
            for partition in partitions:
                offset = self.offsets[partition]
                log.debug('Commit offset %d in SimpleConsumer: '
                          'group=%s, topic=%s, partition=%s',
                          offset, self.group, self.topic, partition)

                reqs.append(OffsetCommitRequest(self.topic, partition,
                                                offset, None))

            try:
                self.client.send_offset_commit_request(self.group, reqs)
            except KafkaError as e:
                log.error('%s saving offsets: %s', e.__class__.__name__, e)
                return False
            else:
                self.count_since_commit = 0
                return True
76629
private void processPartitionResponse(Connection connection, Collection<Map.Entry<Address, List<Integer>>> partitions,
                                          int partitionStateVersion,
                                          boolean partitionStateVersionExist) {

        while (true) {
            PartitionTable current = this.partitionTable.get();
            if (!shouldBeApplied(connection, partitions, partitionStateVersion, partitionStateVersionExist, current)) {
                return;
            }
            Int2ObjectHashMap<Address> newPartitions = convertToPartitionToAddressMap(partitions);
            PartitionTable newMetaData = new PartitionTable(connection, partitionStateVersion, newPartitions);
            if (this.partitionTable.compareAndSet(current, newMetaData)) {
                // partition count is set once at the start. Even if we reset the partition table when switching cluster
                //we want to remember the partition count. That is why it is a different field.
                if (partitionCount == 0) {
                    partitionCount = newPartitions.size();
                }
                if (logger.isFinestEnabled()) {
                    logger.finest("Processed partition response. partitionStateVersion : "
                            + (partitionStateVersionExist ? partitionStateVersion : "NotAvailable")
                            + ", partitionCount :" + newPartitions.size() + ", connection : " + connection);
                }
                return;
            }

        }
    }
1316929
void scheduleActiveMigrationFinalization(final MigrationInfo migrationInfo) {
        partitionServiceLock.lock();
        try {
            if (migrationInfo.equals(activeMigrationInfo)) {
                if (activeMigrationInfo.startProcessing()) {
                    activeMigrationInfo.setStatus(migrationInfo.getStatus());
                    finalizeMigration(activeMigrationInfo);
                } else {
                    // This case happens when master crashes while migration operation is running
                    // and new master publishes the latest completed migrations
                    // after collecting active and completed migrations from existing members.
                    // See FetchMostRecentPartitionTableTask.
                    logger.info("Scheduling finalization of " + migrationInfo
                            + ", because migration process is currently running.");
                    nodeEngine.getExecutionService().schedule(() ->
                            scheduleActiveMigrationFinalization(migrationInfo), 1, TimeUnit.SECONDS);
                }
                return;
            }

            PartitionReplica source = migrationInfo.getSource();
            if (source != null && migrationInfo.getSourceCurrentReplicaIndex() > 0
                    && source.isIdentical(node.getLocalMember())) {
                // Finalize migration on old backup replica owner
                finalizeMigration(migrationInfo);
            }
        } finally {
            partitionServiceLock.unlock();
        }
    }
386578
public boolean checkKeyBelongsToNode(byte[] key, int nodeId) {
        List<Integer> nodePartitions = cluster.getNodeById(nodeId).getPartitionIds();
        List<Integer> replicatingPartitions = getReplicatingPartitionList(key);
        // remove all partitions from the list, except those that belong to the
        // node
        replicatingPartitions.retainAll(nodePartitions);
        return replicatingPartitions.size() > 0;
    }
568774
public synchronized boolean startUpgrade() throws IOException {
    if(upgradeState) {  // upgrade is already in progress
      assert currentUpgrades != null : 
        "UpgradeManagerDatanode.currentUpgrades is null.";
      UpgradeObjectDatanode curUO = (UpgradeObjectDatanode)currentUpgrades.first();
      curUO.startUpgrade();
      return true;
    }
    if(broadcastCommand != null) {
      if(broadcastCommand.getVersion() > this.getUpgradeVersion()) {
        // stop broadcasting, the cluster moved on
        // start upgrade for the next version
        broadcastCommand = null;
      } else {
        // the upgrade has been finished by this data-node,
        // but the cluster is still running it, 
        // reply with the broadcast command
        assert currentUpgrades == null : 
          "UpgradeManagerDatanode.currentUpgrades is not null.";
        assert upgradeDaemon == null : 
          "UpgradeManagerDatanode.upgradeDaemon is not null.";
        dataNode.getNSNamenode(namespaceId).processUpgradeCommand(broadcastCommand);
        return true;
      }
    }
    if(currentUpgrades == null)
      currentUpgrades = getDistributedUpgrades();
    if(currentUpgrades == null) {
      DataNode.LOG.info("\n   Distributed upgrade for DataNode version " 
          + getUpgradeVersion() + " to current LV " 
          + FSConstants.LAYOUT_VERSION + " cannot be started. "
          + "The upgrade object is not defined.");
      return false;
    }
    upgradeState = true;
    UpgradeObjectDatanode curUO = (UpgradeObjectDatanode)currentUpgrades.first();
    curUO.setDatanode(dataNode);
    curUO.startUpgrade();
    upgradeDaemon = new Daemon(curUO);
    upgradeDaemon.start();
    DataNode.LOG.info("\n   Distributed upgrade for DataNode " 
        + dataNode.getDatanodeInfo() 
        + " version " + getUpgradeVersion() + " to current LV " 
        + FSConstants.LAYOUT_VERSION + " is started.");
    return true;
  }
1299326
@VisibleForTesting
	void onPartitionStateUpdate(
			IntermediateDataSetID intermediateDataSetId,
			ResultPartitionID resultPartitionId,
			ExecutionState producerState) throws IOException, InterruptedException {

		if (executionState == ExecutionState.RUNNING) {
			final SingleInputGate inputGate = inputGatesById.get(intermediateDataSetId);

			if (inputGate != null) {
				if (producerState == ExecutionState.SCHEDULED
					|| producerState == ExecutionState.DEPLOYING
					|| producerState == ExecutionState.RUNNING
					|| producerState == ExecutionState.FINISHED) {

					// Retrigger the partition request
					inputGate.retriggerPartitionRequest(resultPartitionId.getPartitionId());

				} else if (producerState == ExecutionState.CANCELING
					|| producerState == ExecutionState.CANCELED
					|| producerState == ExecutionState.FAILED) {

					// The producing execution has been canceled or failed. We
					// don't need to re-trigger the request since it cannot
					// succeed.
					if (LOG.isDebugEnabled()) {
						LOG.debug("Cancelling task {} after the producer of partition {} with attempt ID {} has entered state {}.",
							taskNameWithSubtask,
							resultPartitionId.getPartitionId(),
							resultPartitionId.getProducerId(),
							producerState);
					}

					cancelExecution();
				} else {
					// Any other execution state is unexpected. Currently, only
					// state CREATED is left out of the checked states. If we
					// see a producer in this state, something went wrong with
					// scheduling in topological order.
					String msg = String.format("Producer with attempt ID %s of partition %s in unexpected state %s.",
						resultPartitionId.getProducerId(),
						resultPartitionId.getPartitionId(),
						producerState);

					failExternally(new IllegalStateException(msg));
				}
			} else {
				failExternally(new IllegalStateException("Received partition producer state for " +
						"unknown input gate " + intermediateDataSetId + "."));
			}
		} else {
			LOG.debug("Task {} ignored a partition producer state notification, because it's not running.", taskNameWithSubtask);
		}
	}
1259163
def need_rejoin(self):
        
        if not self._subscription.partitions_auto_assigned():
            return False

        if self._auto_assign_all_partitions():
            return False

        # we need to rejoin if we performed the assignment and metadata has changed
        if (self._assignment_snapshot is not None
            and self._assignment_snapshot != self._metadata_snapshot):
            return True

        # we need to join if our subscription has changed since the last join
        if (self._joined_subscription is not None
            and self._joined_subscription != self._subscription.subscription):
            return True

        return super(ConsumerCoordinator, self).need_rejoin()
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();
        }
    }