Return workers not assigned to any affinity pool at all.

@param affinityConfig affinity config
@param workerMap      map of worker hostname to worker info

@return map of worker hostname to worker info
425489
private static ImmutableMap<String, ImmutableWorkerInfo> getNonAffinityWorkers(
      final AffinityConfig affinityConfig,
      final Map<String, ImmutableWorkerInfo> workerMap
  )
  {
    return ImmutableMap.copyOf(
        Maps.filterKeys(
            workerMap,
            workerHost -> !affinityConfig.getAffinityWorkers().contains(workerHost)
        )
    );
  }
325096
def ConfigureLazyWorkers(self):
    
    lazy_worker_instances = self.__GetMissingWorkers()
    if not lazy_worker_instances:
      return
    reachable_states = self.__AreInstancesReachable(lazy_worker_instances)
    reachable_instances = [t[0] for t in zip(lazy_worker_instances, 
                                             reachable_states) if t[1]]
    print 'reachable_instances: %s' % reachable_instances 
    self.__ConfigureWorkers(reachable_instances)
    return
81905
public static Map<String, List<Interaction>> getPoolData(List<TraceEvent> data,
                                                            boolean ignoreDelist, boolean ignoreTracking,
                                                            boolean ignoreIncomplete)
   {
      // Pool -> Interactions
      Map<String, List<Interaction>> result = new TreeMap<String, List<Interaction>>();
      
      // Pool -> ConnectionListener -> Events
      Map<String, Map<String, List<TraceEvent>>> temp = new TreeMap<String, Map<String, List<TraceEvent>>>();

      for (int i = 0; i < data.size(); i++)
      {
         TraceEvent te = data.get(i);

         Map<String, List<TraceEvent>> m = temp.get(te.getPool());

         if (m == null)
            m = new TreeMap<String, List<TraceEvent>>();

         List<TraceEvent> l = m.get(te.getConnectionListener());

         if (l == null)
            l = new ArrayList<TraceEvent>();

         l.add(te);

         if (isEndState(te))
         {
            Interaction interaction = new Interaction(te.getThreadId(),
                                                      l.get(0).getTimestamp(),
                                                      l.get(l.size() - 1).getTimestamp(),
                                                      l,
                                                      getStatus(l, ignoreDelist, ignoreTracking, ignoreIncomplete));
            List<Interaction> pool = result.get(te.getPool());

            if (pool == null)
               pool = new ArrayList<Interaction>();

            pool.add(interaction);
            result.put(te.getPool(), pool);
            l = null;
         }
         
         m.put(te.getConnectionListener(), l);
         temp.put(te.getPool(), m);
      }

      if (!ignoreIncomplete)
      {
         for (Map.Entry<String, Map<String, List<TraceEvent>>> poolEntry : temp.entrySet())
         {
            for (Map.Entry<String, List<TraceEvent>> clEntry : poolEntry.getValue().entrySet())
            {
               List<TraceEvent> l = clEntry.getValue();

               if (l != null)
               {
                  Interaction interaction = new Interaction(l.get(0).getThreadId(),
                                                            l.get(0).getTimestamp(),
                                                            l.get(l.size() - 1).getTimestamp(),
                                                            l,
                                                            getStatus(l, ignoreDelist, ignoreTracking,
                                                                      ignoreIncomplete));
                  List<Interaction> pool = result.get(poolEntry.getKey());
               
                  if (pool == null)
                     pool = new ArrayList<Interaction>();

                  pool.add(interaction);
                  result.put(poolEntry.getKey(), pool);
               }
            }
         }
      }
      
      return result;
   }
538463
protected Map<String, Object> groupIndexMap(
            final List<DocWorkUnit> workUnitList,
            final List<Map<String, String>> groupMaps
    ) {
        //
        // root -> data -> { summary -> y, filename -> z }, etc
        //      -> groups -> group1, group2, etc.
        Map<String, Object> root = new HashMap<>();

        Collections.sort(workUnitList);

        List<Map<String, String>> data = new ArrayList<>();
        workUnitList.stream().forEach(workUnit -> data.add(indexDataMap(workUnit)));

        root.put("data", data);
        root.put("groups", groupMaps);
        root.put("timestamp", getBuildTimeStamp());
        root.put("version", getBuildVersion());

        return root;
    }
1035087
public String sandboxPolicy(String workerId, Map<String, String> replaceMap) throws IOException {
        if (!isEnable) {
            return "";
        }

        replaceMap.putAll(replaceBaseMap);
        String tmpPolicy = generatePolicyFile(replaceMap);

        File file = new File(tmpPolicy);
        String policyPath = StormConfig.worker_root(conf, workerId) + File.separator + SANBOX_TEMPLATE_NAME;
        File dest = new File(policyPath);
        file.renameTo(dest);

        StringBuilder sb = new StringBuilder();
        sb.append(" -Djava.security.manager -Djava.security.policy=");
        sb.append(policyPath);

        return sb.toString();
    }
274761
def workers(profile='default'):
    '''
    
    '''

    config = get_running(profile)
    lbn = config['worker.list'].split(',')
    worker_list = []
    ret = {}

    for lb in lbn:
        try:
            worker_list.extend(
                config['worker.{0}.balance_workers'.format(lb)].split(',')
            )
        except KeyError:
            pass

    worker_list = list(set(worker_list))

    for worker in worker_list:
        ret[worker] = {
            'activation': config['worker.{0}.activation'.format(worker)],
            'state': config['worker.{0}.state'.format(worker)],
        }

    return ret
792260
@Nullable
  private MasterWorkerInfo findUnregisteredWorker(long workerId) {
    for (IndexedSet<MasterWorkerInfo> workers: Arrays.asList(mTempWorkers, mLostWorkers)) {
      MasterWorkerInfo worker = workers.getFirstByField(ID_INDEX, workerId);
      if (worker != null) {
        return worker;
      }
    }
    return null;
  }
809143
def get_worker_node_ips(config_file, override_cluster_name):
    """"""

    config = yaml.load(open(config_file).read())
    if override_cluster_name is not None:
        config["cluster_name"] = override_cluster_name

    provider = get_node_provider(config["provider"], config["cluster_name"])
    try:
        nodes = provider.non_terminated_nodes({TAG_RAY_NODE_TYPE: "worker"})

        if config.get("provider", {}).get("use_internal_ips", False) is True:
            return [provider.internal_ip(node) for node in nodes]
        else:
            return [provider.external_ip(node) for node in nodes]
    finally:
        provider.cleanup()
368756
@SuppressWarnings("rawtypes")
    public static WorkerShutdown mk_worker(Map conf, IContext context, String topologyId, String supervisorId,
                                           int port, String workerId, String jarPath) throws Exception {
        StringBuilder sb = new StringBuilder();
        sb.append("topologyId:").append(topologyId).append(", ");
        sb.append("port:").append(port).append(", ");
        sb.append("workerId:").append(workerId).append(", ");
        sb.append("jarPath:").append(jarPath).append("\n");

        LOG.info("Begin to run worker:" + sb.toString());
        Worker w = new Worker(conf, context, topologyId, supervisorId, port, workerId, jarPath);
        //   w.redirectOutput();

        return w.execute();
    }
803124
private static String getWorkerFormattedTierValues(Map<String, Map<String, String>> map,
      String workerName) {
    return map.values().stream().map((tierMap)
        -> (String.format("%-14s", tierMap.getOrDefault(workerName, "-"))))
        .collect(Collectors.joining(""));
  }
430962
public Map<String, WorkerHeartbeat> readWorkerHeartbeats(Map conf) throws Exception {
        Map<String, WorkerHeartbeat> workerHeartbeats = new HashMap<>();

        // get the path: STORM-LOCAL-DIR/workers
        String path = StormConfig.worker_root(conf);

        List<String> workerIds = PathUtils.read_dir_contents(path);

        if (workerIds == null) {
            LOG.info("No worker dir under " + path);
            return workerHeartbeats;
        }

        for (String workerId : workerIds) {
            WorkerHeartbeat whb = readWorkerHeartbeat(conf, workerId);
            // ATTENTION: whb can be null
            workerHeartbeats.put(workerId, whb);
        }
        return workerHeartbeats;
    }
4280
public static org.apache.hadoop.mapred.JobConf asJobConf(Configuration cfg) {
        return cfg instanceof org.apache.hadoop.mapred.JobConf
            ? (org.apache.hadoop.mapred.JobConf) cfg
            : new org.apache.hadoop.mapred.JobConf(cfg);
    }
654407
public void checkNewWorkers(Map conf) throws IOException, InterruptedException {
        Set<String> workers = new HashSet<>();
        for (Entry<String, Pair<Integer, Integer>> entry : workerIdToStartTimeAndPort.entrySet()) {
            String workerId = entry.getKey();
            int startTime = entry.getValue().getFirst();
            LocalState ls = StormConfig.worker_state(conf, workerId);
            WorkerHeartbeat whb = (WorkerHeartbeat) ls.get(Common.LS_WORKER_HEARTBEAT);
            if (whb == null) {
                if ((TimeUtils.current_time_secs() - startTime) <
                        JStormUtils.parseInt(conf.get(Config.SUPERVISOR_WORKER_START_TIMEOUT_SECS))) {
                    LOG.info(workerId + " still hasn't started");
                } else {
                    LOG.error("Failed to start Worker " + workerId);
                    workers.add(workerId);
                }
            } else {
                LOG.info("Successfully started worker " + workerId);
                workers.add(workerId);
            }
        }
        for (String workerId : workers) {
            Integer port = this.workerIdToStartTimeAndPort.get(workerId).getSecond();
            this.workerIdToStartTimeAndPort.remove(workerId);
            this.portToWorkerId.remove(port);
        }
    }
364826
def filter_workers(self, request, context):
        """"""
        _log_request(request, context)
        workers_pattern, workers_negate = PATTERN_PARAMS_OP(request.workers_filter)

        hregex = re.compile(workers_pattern)  # hostname filter condition

        def hcondition(worker):
            return accepts(hregex, workers_negate, worker.hostname)  # pragma: no branch

        found_workers = (worker for worker in
                         sorted(self.listener.memory.workers.values(),
                                key=WORKER_HOSTNAME_OP)
                         if hcondition(worker))

        def callback(t):
            logger.debug('%s iterated %d workers in %s (%s)', self.filter_workers.__name__,
                         t.count, t.duration_human, t.throughput_human)

        for worker in about_time(callback, found_workers):
            yield ClearlyServer._event_to_pb(worker)[1]
124352
private String[] getMappedServers(final ServerInformationCatalog serverInformationCatalog,
            final Map<String, String> mappedServers) {

        if (serverInformationCatalog.containsServer(HadoopResource.DEFAULT_CLUSTERREFERENCE)) {
            mappedServers.put("(default)", HadoopResource.DEFAULT_CLUSTERREFERENCE);
        }

        final String[] serverNames = serverInformationCatalog.getServerNames();
        for (int i = 0; i < serverNames.length; i++) {
            final String serverName = serverNames[i];
            if (!serverName.equals(HadoopResource.DEFAULT_CLUSTERREFERENCE)) {
                mappedServers.put(serverName, serverName);
            }
        }
        return mappedServers.keySet().toArray(new String[serverNames.length]);
    }
182793
public static <V> V doWorkInPoolNicely(final Pool<Jedis> pool, final PoolWork<Jedis, V> work) {
        final V result;
        try {
            result = doWorkInPool(pool, work);
        } catch (RuntimeException re) {
            throw re;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }
885278
MultimapWithProtoValuesFluentAssertion<M> usingConfig(FluentEqualityConfig newConfig) {
    Subject.Factory<MultimapWithMessageValuesSubject<K, M>, Multimap<K, M>> factory =
        multimapWithMessageValues(newConfig);
    MultimapWithMessageValuesSubject<K, M> newSubject = check().about(factory).that(actual());
    if (internalCustomName() != null) {
      newSubject = newSubject.named(internalCustomName());
    }
    return new MultimapWithProtoValuesFluentAssertionImpl<>(newSubject);
  }