Creates a collector for a single key write operation.
@param id           the id from {@link CommandInvocationId}.
@param backupOwners the backup owners of the key.
@param topologyId   the current topology id.
485007
public <T> Collector<T> create(long id, Collection<Address> backupOwners, int topologyId) {
      if (backupOwners.isEmpty()) {
         return new PrimaryOwnerOnlyCollector<>();
      }
      SingleKeyCollector<T> collector = new SingleKeyCollector<>(id, backupOwners, topologyId);
      BaseAckTarget prev = collectorMap.put(id, collector);
      //is it possible the have a previous collector when the topology changes after the first collector is created
      //in that case, the previous collector must have a lower topology id
      assert prev == null || prev.topologyId < topologyId : format("replaced old collector '%s' by '%s'", prev, collector);
      if (trace) {
         log.tracef("Created new collector for %s. BackupOwners=%s", id, backupOwners);
      }
      return collector;
   }
1048917
def publish_topology_closed(self, topology_id):
        
        event = TopologyClosedEvent(topology_id)
        for subscriber in self.__topology_listeners:
            try:
                subscriber.closed(event)
            except Exception:
                _handle_exception()
831431
@SuppressWarnings("unchecked")
    public static Map component_conf(TopologyContext topology_context, String component_id) {
        Map<Object, Object> componentConf = new HashMap<>();
        String jconf = topology_context.getComponentCommon(component_id).get_json_conf();
        if (jconf != null) {
            componentConf = (Map<Object, Object>) JStormUtils.from_json(jconf);
        }

        return componentConf;
    }
703014
public static void validate_basic(StormTopology topology, Map<Object, Object> totalStormConf,
                                      String topologyId) throws InvalidTopologyException {
        validate_ids(topology, topologyId);

        for (StormTopology._Fields field : Thrift.SPOUT_FIELDS) {
            Object value = topology.getFieldValue(field);
            if (value != null) {
                Map<String, Object> obj_map = (Map<String, Object>) value;
                for (Object obj : obj_map.values()) {
                    validate_component_inputs(obj);
                }
            }
        }

        Integer workerNum = JStormUtils.parseInt(totalStormConf.get(Config.TOPOLOGY_WORKERS));
        if (workerNum == null || workerNum <= 0) {
            String errMsg = "There is no Config.TOPOLOGY_WORKERS in configuration of " + topologyId;
            throw new InvalidParameterException(errMsg);
        }

        Integer ackerNum = JStormUtils.parseInt(totalStormConf.get(Config.TOPOLOGY_ACKER_EXECUTORS));
        if (ackerNum != null && ackerNum < 0) {
            String errMsg = "Invalid Config.TOPOLOGY_ACKERS in configuration of " + topologyId;
            throw new InvalidParameterException(errMsg);
        }

    }
709495
def failback_from_replicant(self, volume_id, replicant_id):
        

        return self.client.call('Network_Storage', 'failbackFromReplicant',
                                replicant_id, id=volume_id)
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;
        }
    }
79934
public void setBaseId(byte[] baseId) {
    if (baseId == null) {
      throw new IllegalArgumentException("Transmitter base ID cannot be null.");
    }
    if (baseId.length != TRANSMITTER_ID_SIZE) {
      throw new IllegalArgumentException(String.format(
          "Transmitter base ID must be %d bytes long.",
          Integer.valueOf(TRANSMITTER_ID_SIZE)));
    }
    this.baseId = baseId;
  }
1123172
public void setRelationId(String relationId) {

        try {
            m_relationId = new CmsUUID(relationId);
        } catch (Throwable e) {
            setThrowable(e);
        }
    }
476001
synchronized public void prepareTransferMastership(int partitionId, int hostId) {
        // remove mastership for partition id, so when failure happen during the mastership transfer
        // this node can be elected as new master again.
        m_masterOfPartitions.remove(partitionId);

        if (exportLog.isDebugEnabled()) {
            exportLog.debug("Export stream masters on " + partitionId + " are going to migrate away");
        }
        ExportGeneration generation = m_generation.get();
        if (generation == null) {
            return;
        }
        generation.prepareTransferMastership(partitionId, hostId);
    }
322663
void scheduleOrUpdateConsumers(ResultPartitionID partitionId) {

		final Execution execution = currentExecution;

		// Abort this request if there was a concurrent reset
		if (!partitionId.getProducerId().equals(execution.getAttemptId())) {
			return;
		}

		final IntermediateResultPartition partition = resultPartitions.get(partitionId.getPartitionId());

		if (partition == null) {
			throw new IllegalStateException("Unknown partition " + partitionId + ".");
		}

		partition.markDataProduced();

		if (partition.getIntermediateResult().getResultType().isPipelined()) {
			// Schedule or update receivers of this partition
			execution.scheduleOrUpdateConsumers(partition.getConsumers());
		}
		else {
			throw new IllegalArgumentException("ScheduleOrUpdateConsumers msg is only valid for" +
					"pipelined partitions.");
		}
	}
344641
@SuppressWarnings("unchecked")
    public <T extends WebLocatorAbstractBuilder> T setId(final String id) {
        pathBuilder.setId(id);
        return (T) this;
    }
715436
@Override
  public final boolean setRoutingObjectID(DBID objectID) {
    if(objectID == routingObjectID || DBIDUtil.equal(objectID, routingObjectID)) {
      return false;
    }
    this.routingObjectID = objectID;
    return true;
  }
1087680
public void setHostKeys(java.util.Collection<HostKeyAttributes> hostKeys) {
        if (hostKeys == null) {
            this.hostKeys = null;
            return;
        }

        this.hostKeys = new java.util.ArrayList<HostKeyAttributes>(hostKeys);
    }
183733
private SupervisorWorkers getSupervisorWorkersByHostOrId(String host, String id) throws TException {
        long start = System.nanoTime();

        if (StringUtils.isBlank(id) && StringUtils.isBlank(host)) {
            throw new TException("Must specify host or supervisor id!");
        }

        try {
            StormClusterState stormClusterState = data.getStormClusterState();

            // all supervisors
            Map<String, SupervisorInfo> supervisorInfos = Cluster.get_all_SupervisorInfo(stormClusterState, null);
            SupervisorInfo supervisorInfo = null;
            String ip;
            if (!StringUtils.isBlank(id)) {
                supervisorInfo = supervisorInfos.get(id);
                host = supervisorInfo.getHostName();
                ip = NetWorkUtils.host2Ip(host);
            } else {
                ip = NetWorkUtils.host2Ip(host);
                for (Entry<String, SupervisorInfo> entry : supervisorInfos.entrySet()) {
                    SupervisorInfo info = entry.getValue();
                    if (info.getHostName().equals(host) || info.getHostName().equals(ip)) {
                        id = entry.getKey();
                        supervisorInfo = info;
                        break;
                    }
                }
            }
            if (supervisorInfo == null) {
                throw new TException("unknown supervisor id:" + id);
            }

            Map<String, Assignment> assignments = Cluster.get_all_assignment(stormClusterState, null);
            Map<Integer, WorkerSummary> portWorkerSummaries = new TreeMap<>();
            int usedSlotNumber = 0;
            Map<String, Map<Integer, String>> topologyTaskToComponent = new HashMap<>();

            Map<String, MetricInfo> metricInfoMap = new HashMap<>();
            for (Entry<String, Assignment> entry : assignments.entrySet()) {
                String topologyId = entry.getKey();
                Assignment assignment = entry.getValue();

                Set<ResourceWorkerSlot> workers = assignment.getWorkers();
                for (ResourceWorkerSlot worker : workers) {
                    if (!id.equals(worker.getNodeId())) {
                        continue;
                    }
                    usedSlotNumber++;

                    Integer port = worker.getPort();
                    WorkerSummary workerSummary = portWorkerSummaries.get(port);
                    if (workerSummary == null) {
                        workerSummary = new WorkerSummary();
                        workerSummary.set_port(port);
                        workerSummary.set_topology(topologyId);
                        workerSummary.set_tasks(new ArrayList<TaskComponent>());

                        portWorkerSummaries.put(port, workerSummary);
                    }

                    Map<Integer, String> taskToComponent = topologyTaskToComponent.get(topologyId);
                    if (taskToComponent == null) {
                        taskToComponent = Cluster.get_all_task_component(stormClusterState, topologyId, null);
                        topologyTaskToComponent.put(topologyId, taskToComponent);
                    }

                    int earliest = TimeUtils.current_time_secs();
                    for (Integer taskId : worker.getTasks()) {
                        TaskComponent taskComponent = new TaskComponent();
                        taskComponent.set_component(taskToComponent.get(taskId));
                        taskComponent.set_taskId(taskId);
                        Integer startTime = assignment.getTaskStartTimeSecs().get(taskId);
                        if (startTime != null && startTime < earliest) {
                            earliest = startTime;
                        }

                        workerSummary.add_to_tasks(taskComponent);
                    }

                    workerSummary.set_uptime(TimeUtils.time_delta(earliest));

                    String workerSlotName = getWorkerSlotName(ip, port);
                    List<MetricInfo> workerMetricInfoList = this.data.getMetricCache().getMetricData(topologyId, MetaType.WORKER);
                    if (workerMetricInfoList.size() > 0) {
                        MetricInfo workerMetricInfo = workerMetricInfoList.get(0);
                        // remove metrics that don't belong to current worker
                        for (Iterator<String> itr = workerMetricInfo.get_metrics().keySet().iterator();
                             itr.hasNext(); ) {
                            String metricName = itr.next();
                            if (!metricName.contains(ip)) {
                                itr.remove();
                            }
                        }
                        metricInfoMap.put(workerSlotName, workerMetricInfo);
                    }
                }
            }

            List<WorkerSummary> workerList = new ArrayList<>();
            workerList.addAll(portWorkerSummaries.values());

            Map<String, Integer> supervisorToUsedSlotNum = new HashMap<>();
            supervisorToUsedSlotNum.put(id, usedSlotNumber);
            SupervisorSummary supervisorSummary = NimbusUtils.mkSupervisorSummary(supervisorInfo, id, supervisorToUsedSlotNum);

            return new SupervisorWorkers(supervisorSummary, workerList, metricInfoMap);
        } catch (TException e) {
            LOG.info("Failed to get ClusterSummary ", e);
            throw e;
        } catch (Exception e) {
            LOG.info("Failed to get ClusterSummary ", e);
            throw new TException(e);
        } finally {
            long end = System.nanoTime();
            SimpleJStormMetric.updateNimbusHistogram("getSupervisorWorkers", (end - start) / TimeUtils.NS_PER_US);
        }

    }
931571
private Object handleNonTxWriteCommand(InvocationContext ctx, WriteCommand command) {
      if (trace) log.tracef("handleNonTxWriteCommand for command %s, topology id %d", command, command.getTopologyId());

      updateTopologyId(command);

      // Only catch OutdatedTopologyExceptions on the originator
      if (!ctx.isOriginLocal()) {
         return invokeNext(ctx, command);
      }

      return invokeNextAndHandle(ctx, command, handleNonTxWriteReturn);
   }
345529
public void restartTopology(String stormId){
        String currentTopology = StringUtils.substringBefore(stormId, "-");
        String topologyPrefix = StringUtils.substringBefore(currentTopology, "_");
        String topologyIndex =  StringUtils.substringAfter(currentTopology, "_");
        int newIndex = Integer.parseInt(topologyIndex) + 1;
        String newTopologyName = topologyPrefix + "_" + newIndex;
        LOGGER.info("new topology name = " + newTopologyName);
        execStartCommand(newTopologyName);
        LOGGER.info("execStartCommand finish ..");
        execShutdownCommand(currentTopology);
        LOGGER.info("execShutdownCommand finish ..");
    }
1243597
@Override
   public void stopApplyingState(int topologyId) {
      if (trace) log.tracef("Stop keeping track of changed keys for state transfer in topology %d", topologyId);
      commitManager.stopTrack(PUT_FOR_STATE_TRANSFER);
   }