Acknowledges a write operation completion in the backup owner.

@param id         the id from {@link CommandInvocationId#getId()}.
@param from       the backup owner.
@param topologyId the topology id.
498046
public void backupAck(long id, Address from, int topologyId) {
      BaseAckTarget ackTarget = collectorMap.get(id);
      if (ackTarget instanceof SingleKeyCollector) {
         ((SingleKeyCollector) ackTarget).backupAck(topologyId, from);
      } else if (ackTarget instanceof MultiTargetCollectorImpl) {
         ((MultiTargetCollectorImpl) ackTarget).backupAck(topologyId, from);
      }
   }
92587
public void shard(ConceptId conceptId) {
        ConceptImpl type = getConcept(conceptId);
        if (type == null) {
            LOG.warn("Cannot shard concept [" + conceptId + "] due to it not existing in the graph");
        } else {
            type.createShard();
        }
    }
23941
public void write(Command cmd)
        throws IOException, IllegalArgumentException {
        //we delete the initial reply when the first command is sent
        if (cmd == null) {
            throw new IllegalArgumentException("null argument: cmd");
        }
        if (logger.isDebugEnabled()) {
            logger.debug("Control channel sending: " + cmd);
        }
        writeStr(cmd.toString());
    }
1078266
public void addObjectId( Object object, ObjectIdSerializer<?> id ) {
        if ( null == mapObjectId ) {
            if ( useEqualityForObjectId ) {
                mapObjectId = new HashMap<Object, ObjectIdSerializer<?>>();
            } else {
                mapObjectId = new IdentityHashMap<Object, ObjectIdSerializer<?>>();
            }
        }
        mapObjectId.put( object, id );
    }
617301
public static Map read_supervisor_topology_conf(Map conf, String topologyId) throws IOException {
        String topologyRoot = StormConfig.supervisor_stormdist_root(conf, topologyId);
        String confPath = StormConfig.stormconf_path(topologyRoot);
        return (Map) readLocalObject(topologyId, confPath);
    }
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;
    }
883685
@Override
    public void setID(String id){
        if (isFrozen()) {
            throw new UnsupportedOperationException("Attempt to modify a frozen OlsonTimeZone instance.");
        }

        // Before updating the ID, preserve the original ID's canonical ID.
        if (canonicalID == null) {
            canonicalID = getCanonicalID(getID());
            assert(canonicalID != null);
            if (canonicalID == null) {
                // This should never happen...
                canonicalID = getID();
            }
        }

        if (finalZone != null){
            finalZone.setID(id);
        }
        super.setID(id);
        transitionRulesInitialized = false;
    }
220058
public void setID(String ID) {
        if (ID == null) {
            throw new NullPointerException();
        }
        if (isFrozen()) {
            throw new UnsupportedOperationException("Attempt to modify a frozen TimeZone instance.");
        }
        this.ID = ID;
    }
17327
private void setId(String id) {
        requireArgument(id != null && !id.trim().isEmpty(), "ID cannot be null or empty.");
        _id = id;
    }
758810
private synchronized void doNodeLeft(final String nodeID) {
    log.info(String.format("%s - %s left the cluster", this, nodeID));
    context.run(new Runnable() {
      @Override
      public void run() {
        // Redeploy any failed deployments.
        synchronized (deployments) {
          Collection<String> sdeploymentsInfo = deployments.get(group);
          for (final String sdeploymentInfo : sdeploymentsInfo) {
            final JsonObject deploymentInfo = new JsonObject(sdeploymentInfo);
            // If the deployment node is equal to the node that left the cluster then
            // remove the deployment from the deployments list and attempt to redeploy it.
            if (deploymentInfo.getString("node").equals(nodeID)) {
              // If the deployment is an HA deployment then attempt to redeploy it on this node.
              if (deployments.remove(group, sdeploymentInfo) && deploymentInfo.getBoolean("ha", false)) {
                doRedeploy(deploymentInfo);
              }
            }
          }
        }
      }
    });
  }
1030860
@Override
    public StormTopology getTopology(String id) throws TException {
        StormTopology topology;
        try {
            StormTopology stormtopology = StormConfig.read_nimbus_topology_code(id, data.getBlobStore());
            if (stormtopology == null) {
                throw new NotAliveException("No topology of " + id);
            }

            Map<Object, Object> topologyConf = (Map<Object, Object>) StormConfig.read_nimbus_topology_conf(id, data.getBlobStore());
            topology = Common.system_topology(topologyConf, stormtopology);
        } catch (Exception e) {
            LOG.error("Failed to get topology " + id + ",", e);
            throw new TException("Failed to get system_topology");
        }
        return topology;
    }
1035925
@Override
    public TopologyInfo getTopologyInfo(String topologyId) throws TException {
        long start = System.nanoTime();
        StormClusterState stormClusterState = data.getStormClusterState();

        try {
            // get topology's StormBase
            StormBase base = stormClusterState.storm_base(topologyId, null);
            if (base == null) {
                throw new NotAliveException("No topology of " + topologyId);
            }

            Assignment assignment = stormClusterState.assignment_info(topologyId, null);
            if (assignment == null) {
                throw new NotAliveException("No topology of " + topologyId);
            }

            TopologyTaskHbInfo topologyTaskHbInfo = data.getTasksHeartbeat().get(topologyId);
            Map<Integer, TaskHeartbeat> taskHbMap = null;
            if (topologyTaskHbInfo != null)
                taskHbMap = topologyTaskHbInfo.get_taskHbs();

            Map<Integer, TaskInfo> taskInfoMap = Cluster.get_all_taskInfo(stormClusterState, topologyId);
            Map<Integer, String> taskToComponent = Cluster.get_all_task_component(stormClusterState, topologyId, taskInfoMap);
            Map<Integer, String> taskToType = Cluster.get_all_task_type(stormClusterState, topologyId, taskInfoMap);


            String errorString;
            if (Cluster.is_topology_exist_error(stormClusterState, topologyId)) {
                errorString = "Y";
            } else {
                errorString = "";
            }

            TopologySummary topologySummary = new TopologySummary();
            topologySummary.set_id(topologyId);
            topologySummary.set_name(base.getStormName());
            topologySummary.set_uptimeSecs(TimeUtils.time_delta(base.getLanchTimeSecs()));
            topologySummary.set_status(base.getStatusString());
            topologySummary.set_numTasks(NimbusUtils.getTopologyTaskNum(assignment));
            topologySummary.set_numWorkers(assignment.getWorkers().size());
            topologySummary.set_errorInfo(errorString);

            Map<String, ComponentSummary> componentSummaryMap = new HashMap<>();
            HashMap<String, List<Integer>> componentToTasks = JStormUtils.reverse_map(taskToComponent);
            for (Entry<String, List<Integer>> entry : componentToTasks.entrySet()) {
                String name = entry.getKey();
                List<Integer> taskIds = entry.getValue();
                if (taskIds == null || taskIds.size() == 0) {
                    LOG.warn("No task of component " + name);
                    continue;
                }

                ComponentSummary componentSummary = new ComponentSummary();
                componentSummaryMap.put(name, componentSummary);

                componentSummary.set_name(name);
                componentSummary.set_type(taskToType.get(taskIds.get(0)));
                componentSummary.set_parallel(taskIds.size());
                componentSummary.set_taskIds(taskIds);
            }

            Map<Integer, TaskSummary> taskSummaryMap = new TreeMap<>();
            Map<Integer, List<TaskError>> taskErrors = Cluster.get_all_task_errors(stormClusterState, topologyId);

            for (Integer taskId : taskInfoMap.keySet()) {
                TaskSummary taskSummary = new TaskSummary();
                taskSummaryMap.put(taskId, taskSummary);

                taskSummary.set_taskId(taskId);
                if (taskHbMap == null) {
                    taskSummary.set_status("Starting");
                    taskSummary.set_uptime(0);
                } else {
                    TaskHeartbeat hb = taskHbMap.get(taskId);
                    if (hb == null) {
                        taskSummary.set_status("Starting");
                        taskSummary.set_uptime(0);
                    } else {
                        boolean isInactive = NimbusUtils.isTaskDead(data, topologyId, taskId);
                        if (isInactive)
                            taskSummary.set_status("INACTIVE");
                        else
                            taskSummary.set_status("ACTIVE");
                        taskSummary.set_uptime(hb.get_uptime());
                    }
                }

                if (StringUtils.isBlank(errorString)) {
                    continue;
                }

                List<TaskError> taskErrorList = taskErrors.get(taskId);
                if (taskErrorList != null && taskErrorList.size() != 0) {
                    for (TaskError taskError : taskErrorList) {
                        ErrorInfo errorInfo = new ErrorInfo(taskError.getError(), taskError.getTimSecs(),
                                taskError.getLevel(), taskError.getCode());
                        taskSummary.add_to_errors(errorInfo);
                        String component = taskToComponent.get(taskId);
                        componentSummaryMap.get(component).add_to_errors(errorInfo);
                    }
                }
            }

            for (ResourceWorkerSlot workerSlot : assignment.getWorkers()) {
                String hostname = workerSlot.getHostname();
                int port = workerSlot.getPort();

                for (Integer taskId : workerSlot.getTasks()) {
                    TaskSummary taskSummary = taskSummaryMap.get(taskId);
                    taskSummary.set_host(hostname);
                    taskSummary.set_port(port);
                }
            }

            TopologyInfo topologyInfo = new TopologyInfo();
            topologyInfo.set_topology(topologySummary);
            topologyInfo.set_components(JStormUtils.mk_list(componentSummaryMap.values()));
            topologyInfo.set_tasks(JStormUtils.mk_list(taskSummaryMap.values()));

            // return topology metric & component metric only
            List<MetricInfo> tpMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.TOPOLOGY);
            List<MetricInfo> compMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.COMPONENT);
            List<MetricInfo> workerMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.WORKER);
            List<MetricInfo> compStreamMetricList = data.getMetricCache().getMetricData(topologyId, MetaType.COMPONENT_STREAM);

            MetricInfo taskMetric = MetricUtils.mkMetricInfo();
            MetricInfo streamMetric = MetricUtils.mkMetricInfo();
            MetricInfo nettyMetric = MetricUtils.mkMetricInfo();
            MetricInfo tpMetric, compMetric, compStreamMetric, workerMetric;

            if (tpMetricList == null || tpMetricList.size() == 0) {
                tpMetric = MetricUtils.mkMetricInfo();
            } else {
                // get the last min topology metric
                tpMetric = tpMetricList.get(tpMetricList.size() - 1);
            }
            if (compMetricList == null || compMetricList.size() == 0) {
                compMetric = MetricUtils.mkMetricInfo();
            } else {
                compMetric = compMetricList.get(0);
            }
            if (compStreamMetricList == null || compStreamMetricList.size() == 0) {
                compStreamMetric = MetricUtils.mkMetricInfo();
            } else {
                compStreamMetric = compStreamMetricList.get(0);
            }
            if (workerMetricList == null || workerMetricList.size() == 0) {
                workerMetric = MetricUtils.mkMetricInfo();
            } else {
                workerMetric = workerMetricList.get(0);
            }
            TopologyMetric topologyMetrics = new TopologyMetric(tpMetric, compMetric, workerMetric,
                    taskMetric, streamMetric, nettyMetric);
            topologyMetrics.set_compStreamMetric(compStreamMetric);
            topologyInfo.set_metrics(topologyMetrics);

            return topologyInfo;
        } catch (TException e) {
            LOG.info("Failed to get topologyInfo " + topologyId, e);
            throw e;
        } catch (Exception e) {
            LOG.info("Failed to get topologyInfo " + topologyId, e);
            throw new TException("Failed to get topologyInfo" + topologyId);
        } finally {
            long end = System.nanoTime();
            SimpleJStormMetric.updateNimbusHistogram("getTopologyInfo", (end - start) / TimeUtils.NS_PER_US);
        }

    }
322880
@POST
    @Path("close")    
    public void close(String id) {
    	Statement statement = StatementHolder.get().get(id);
    	try {
			statement.close();
			StatementHolder.get().remove(id);
		} catch (SQLException e) {
			// TODO
			throw new RuntimeException(e);
		}		
    }
786080
@Override
    public void backup(long id, long version) {
        if (LOG.isDebugEnabled())
            LOG.debug("Backup: {} ver: {} {}", new Object[]{hex(id), version, copyImmediately ? "(COPY)" : ""});
        if (copyImmediately) {
            currentBackups.put(id, makeBackup(cache.getLine(id), version));
            oldMap().remove(id);
        }
        else
            map.put(id, new BackupEntry(id, version));
    }
592912
public Boolean setExecutionState(
      ExecutionEnvironment.ExecutionState executionState, String topologyName) {
    return awaitResult(delegate.setExecutionState(executionState, topologyName));
  }
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);
   }
110821
@Override
  public boolean onRestart(Scheduler.RestartTopologyRequest request) {
    // Containers would be restarted automatically once we destroy it
    int containerId = request.getContainerIndex();

    List<Process> processesToRestart = new LinkedList<>();

    if (containerId == -1) {
      LOG.info("Command to restart the entire topology: " + LocalContext.topologyName(config));
      processesToRestart.addAll(processToContainer.keySet());
    } else {
      // restart that particular container
      LOG.info("Command to restart a container of topology: " + LocalContext.topologyName(config));
      LOG.info("Restart container requested: " + containerId);

      // locate the container and destroy it
      for (Process p : processToContainer.keySet()) {
        if (containerId == processToContainer.get(p)) {
          processesToRestart.add(p);
        }
      }
    }

    if (processesToRestart.isEmpty()) {
      LOG.severe("Container not exist.");
      return false;
    }

    for (Process process : processesToRestart) {
      process.destroy();
    }

    return true;
  }