Java 类com.hazelcast.core.Member 实例源码

/**
 * 
 */
private void createHazelcastInstance()
{
  hazelcast = Hazelcast.getOrCreateHazelcastInstance(hzConfig);
   Set<Member> members = hazelcast.getCluster().getMembers();

   int memberIdCnt = 0;
   for(Member m : members)
   {

     if(m.getStringAttribute(Configurator.NODE_INSTANCE_ID).equals(getInstanceId()))
     {
       memberIdCnt++;
     }
     if(memberIdCnt >= 2){
       stop();
       throw new IllegalStateException("Instance not allowed to join cluster as ["+getInstanceId()+"]. Duplicate name!");
     }

   }
   log.info("** Instance ["+getInstanceId()+"] joined cluster ["+hzConfig.getGroupConfig().getName()+"] successfully **");
}

@Override
public void distributedPropertyDidChange(Member member, Integer oldValue, Integer newValue)
{
    if (newValue == null)
    {
        return;
    }
    if (sneyNode instanceof TrainTracksApp)
    {
        oldValue = ((TrainTracksApp) sneyNode).getNumOfCorners();
        if (oldValue < newValue)
        {
            for (int i = 0; i < (newValue - oldValue); i++)
            {
                ((TrainTracksApp) sneyNode).addCorner();
            }
        }
        else if ((oldValue > newValue) && (newValue >= 0))
        {
            for (int i = 0; i < (oldValue - newValue); i++)
            {
                ((TrainTracksApp) sneyNode).removeCorner();
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member member, Integer oldValue, Integer newValue)
{
    if (newValue == null)
    {
        return;
    }
    if (sneyNode instanceof TrainTracksApp)
    {
        oldValue = ((TrainTracksApp) sneyNode).getNumOfCrosses();
        if (oldValue < newValue)
        {
            for (int i = 0; i < (newValue - oldValue); i++)
            {
                ((TrainTracksApp) sneyNode).addCross();
            }
        }
        else if ((oldValue > newValue) && (newValue >= 0))
        {
            for (int i = 0; i < (oldValue - newValue); i++)
            {
                ((TrainTracksApp) sneyNode).removeCross();
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member m, PerformActionMessage oldValue, PerformActionMessage newValue)
{
    if (!newValue.messageAlreadyReceived())
    {
        if (sneyNode instanceof TrainTracksApp)
        {
            if (newValue.getMessageState() == MESSAGESTATE.ACTIVATE)
            {
                ((TrainTracksApp) sneyNode).setMode(true);
            }
            else
            {
                ((TrainTracksApp) sneyNode).setMode(false);
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member member, PerformActionMessage oldValue, PerformActionMessage newValue)
{
    if (!newValue.messageAlreadyReceived())
    {
        if (sneyNode instanceof EnvironmentExplorerApp)
        {
            if (newValue.getMessageState() == MESSAGESTATE.ACTIVATE)
            {
                ((EnvironmentExplorerApp) sneyNode).setTeacherControlVisibility(true);
            }
            else
            {
                ((EnvironmentExplorerApp) sneyNode).setTeacherControlVisibility(false);
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member member, Integer oldValue, Integer newValue)
{
    if (newValue == null)
    {
        return;
    }
    if (sneyNode instanceof TrainTracksApp)
    {
        oldValue = ((TrainTracksApp) sneyNode).getNumOfStraights();
        if (oldValue < newValue)
        {
            for (int i = 0; i < (newValue - oldValue); i++)
            {
                ((TrainTracksApp) sneyNode).addStraight();
            }
        }
        else if ((oldValue > newValue) && (newValue >= 0))
        {
            for (int i = 0; i < (oldValue - newValue); i++)
            {
                ((TrainTracksApp) sneyNode).removeStraight();
            }
        }
    }
}

/**
 * Inits the calculator collection sync.
 */
private void initCalculatorCollectionSync()
{
    List<String> participants = getNamesListFromParticipantList(tableClusterData.getParticipantListControlVariable().getValue());
    calculatorCollectionManager.setAvailableCalculatorCollection(participants);

    tableClusterData.getParticipantListControlVariable().registerChangeListener(new DistributedPropertyChangedAction<List<Participant>>()
    {
        @Override
        public void distributedPropertyDidChange(Member m, List<Participant> oldValue, List<Participant> newValue)
        {
            log.fine("Update received for participant list: " + newValue);
            List<String> names = getNamesListFromParticipantList(newValue);
            calculatorCollectionManager.setAvailableCalculatorCollection(names);
        }
    });
}

/**
 * Instantiates a new expression session.
 *
 * @param dataCluster
 *            the data cluster
 */
public ExpressionSession(NumberNetStudentTableClusteredData dataCluster)
{
    listeners = new ArrayList<IExpressionSessionChangeListener>();

    targetValueFormatter = new DecimalFormat();
    targetValueFormatter.setDecimalSeparatorAlwaysShown(false);
    targetValueFormatter.setMaximumFractionDigits(10);

    expressions = new ArrayList<Expression>();

    dataCluster.getTargetValueControlVariable().registerChangeListener(new DistributedPropertyChangedAction<Double>()
    {
        @Override
        public void distributedPropertyDidChange(Member member, Double oldValue, Double newValue)
        {
            log.fine("Target value changed from " + oldValue + " to " + newValue);
            setTarget(newValue);
        }
    });
}

/**
 * Creates the and get item added action.
 *
 * @return the item added action
 */
private ItemAddedAction<String, Expression> createAndGetItemAddedAction()
{
    itemAddedAction = new ItemAddedAction<String, Expression>()
    {
        @Override
        public void itemAddedToCollection(IMap<String, Expression> collection, String itemKey, Expression itemValue, Member member)
        {
            if (member.localMember())
            {
                return;
            }
            addExpressionFromAlreadyExistingEntryInDistributedMap(itemValue);
        }
    };
    return itemAddedAction;
}

/**
 * Creates the and get item removed action.
 *
 * @return the item removed action
 */
private ItemRemovedAction<String, Expression> createAndGetItemRemovedAction()
{
    itemRemovedAction = new ItemRemovedAction<String, Expression>()
    {
        @Override
        public void itemRemovedFromCollection(IMap<String, Expression> collection, String itemKey, Expression itemValue, Member member)
        {
            if (member.localMember())
            {
                return;
            }
            removeExpressionFromSession(itemValue, true);
        }
    };
    return itemRemovedAction;
}

@Override
public void distributedPropertyDidChange(Member member, Integer oldValue, Integer newValue)
{
    if (newValue == null)
    {
        return;
    }
    if (sneyNode instanceof TrainTracksTrackedApp)
    {
        oldValue = ((TrainTracksTrackedApp) sneyNode).getNumOfCorners();
        if (oldValue < newValue)
        {
            for (int i = 0; i < (newValue - oldValue); i++)
            {
                ((TrainTracksTrackedApp) sneyNode).addCorner();
            }
        }
        else if ((oldValue > newValue) && (newValue >= 0))
        {
            for (int i = 0; i < (oldValue - newValue); i++)
            {
                ((TrainTracksTrackedApp) sneyNode).removeCorner();
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member member, Integer oldValue, Integer newValue)
{
    if (newValue == null)
    {
        return;
    }
    if (sneyNode instanceof TrainTracksTrackedApp)
    {
        oldValue = ((TrainTracksTrackedApp) sneyNode).getNumOfCrosses();
        if (oldValue < newValue)
        {
            for (int i = 0; i < (newValue - oldValue); i++)
            {
                ((TrainTracksTrackedApp) sneyNode).addCross();
            }
        }
        else if ((oldValue > newValue) && (newValue >= 0))
        {
            for (int i = 0; i < (oldValue - newValue); i++)
            {
                ((TrainTracksTrackedApp) sneyNode).removeCross();
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member m, PerformActionMessage oldValue, PerformActionMessage newValue)
{
    if (!newValue.messageAlreadyReceived())
    {
        if (sneyNode instanceof TrainTracksTrackedApp)
        {
            if (newValue.getMessageState() == MESSAGESTATE.ACTIVATE)
            {
                ((TrainTracksTrackedApp) sneyNode).setMode(true);
            }
            else
            {
                ((TrainTracksTrackedApp) sneyNode).setMode(false);
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member member, PerformActionMessage oldValue, PerformActionMessage newValue)
{
    if (!newValue.messageAlreadyReceived())
    {
        if (sneyNode instanceof EnvironmentExplorerTrackedApp)
        {
            if (newValue.getMessageState() == MESSAGESTATE.ACTIVATE)
            {
                ((EnvironmentExplorerTrackedApp) sneyNode).setTeacherControlVisibility(true);
            }
            else
            {
                ((EnvironmentExplorerTrackedApp) sneyNode).setTeacherControlVisibility(false);
            }
        }
    }
}

@Override
public void distributedPropertyDidChange(Member member, Integer oldValue, Integer newValue)
{
    if (newValue == null)
    {
        return;
    }
    if (sneyNode instanceof TrainTracksTrackedApp)
    {
        oldValue = ((TrainTracksTrackedApp) sneyNode).getNumOfStraights();
        if (oldValue < newValue)
        {
            for (int i = 0; i < (newValue - oldValue); i++)
            {
                ((TrainTracksTrackedApp) sneyNode).addStraight();
            }
        }
        else if ((oldValue > newValue) && (newValue >= 0))
        {
            for (int i = 0; i < (oldValue - newValue); i++)
            {
                ((TrainTracksTrackedApp) sneyNode).removeStraight();
            }
        }
    }
}

/**
 * Submits a task to multiple members of the Hazelcast task cluster.
 * 
 * @param task
 *        The task
 * @param where
 *        A {@link MemberSelector}, an {@link Iterable} of {@link Member}, a
 *        string (comma-separated member tags), or null for all members
 * @return A map of members to futures for the task
 */
@SuppressWarnings("unchecked")
private <T> Map<Member, Future<T>> multiTask( SerializableApplicationTask<T> task, Object where )
{
    IExecutorService executor = getHazelcastExecutorService();

    if( where instanceof String )
        return executor.submitToMembers( task, new TaggedMembers( HAZELCAST_MEMBER_TAGS_ATTRIBUTE, (String) where ) );
    else if( where instanceof MemberSelector )
        return executor.submitToMembers( task, (MemberSelector) where );
    else if( where instanceof Collection )
        return executor.submitToMembers( task, (Collection<Member>) where );
    else if( where instanceof Iterable )
    {
        ArrayList<Member> members = new ArrayList<Member>();
        for( Member member : (Iterable<Member>) where )
            members.add( member );
        return executor.submitToMembers( task, members );
    }
    else
        return executor.submitToAllMembers( task );
}

private Member getTaskMember( String tagsAttribute, String requiredTags )
{
    String[] requiredTagsArray = requiredTags.split( "," );
    Set<String> requiredTagsSet = new HashSet<String>();
    for( String tag : requiredTagsArray )
        requiredTagsSet.add( tag );

    for( Member member : getHazelcastTaskInstance().getCluster().getMembers() )
    {
        String tags = (String) member.getAttributes().get( tagsAttribute );
        if( tags != null )
        {
            String[] existingsTags = tags.split( "," );
            for( String existingTag : existingsTags )
                if( requiredTagsSet.contains( existingTag ) )
                    return member;
        }
    }

    return null;
}

@Test
public void testBackpressure() throws Exception {
    DAG dag = new DAG();

    final int member1Port = jet1.getCluster().getLocalMember().getAddress().getPort();
    final Member member2 = jet2.getCluster().getLocalMember();
    final int ptionOwnedByMember2 =
            jet1.getHazelcastInstance().getPartitionService()
                .getPartitions().stream()
                .filter(p -> p.getOwner().equals(member2))
                .map(Partition::getPartitionId)
                .findAny()
                .orElseThrow(() -> new RuntimeException("Can't find a partition owned by member " + jet2));
    Vertex source = dag.newVertex("source", ProcessorMetaSupplier.of((Address address) ->
            ProcessorSupplier.of(address.getPort() == member1Port ? GenerateP::new : noopP())
    ));
    Vertex hiccup = dag.newVertex("hiccup", HiccupP::new);
    Vertex sink = dag.newVertex("sink", SinkProcessors.writeMapP("counts"));

    dag.edge(between(source, hiccup)
            .distributed().partitioned(wholeItem(), (x, y) -> ptionOwnedByMember2))
       .edge(between(hiccup, sink));

    jet1.newJob(dag).join();
    assertCounts(jet1.getMap("counts"));
}

@Override
public void init(@Nonnull Context context) {
    logger = context.jetInstance().getHazelcastInstance().getLoggingService().getLogger(ReadHdfsP.class);
    try {
        int totalParallelism = context.totalParallelism();
        InputFormat inputFormat = jobConf.getInputFormat();
        InputSplit[] splits = inputFormat.getSplits(jobConf, totalParallelism);
        IndexedInputSplit[] indexedInputSplits = new IndexedInputSplit[splits.length];
        Arrays.setAll(indexedInputSplits, i -> new IndexedInputSplit(i, splits[i]));

        Address[] addrs = context.jetInstance().getCluster().getMembers()
                .stream().map(Member::getAddress).toArray(Address[]::new);
        assigned = assignSplitsToMembers(indexedInputSplits, addrs);
        printAssignments(assigned);
    } catch (IOException e) {
        throw rethrow(e);
    }
}

private void test(final HazelcastInstance hazelcast1, final HazelcastInstance hazelcast2) {
    final IAtomicLong counterA = hazelcast1.getAtomicLong("counter-a");
    final long countA = counterA.incrementAndGet();
    LOG.log(INFO, "expecting {0} = {1}", new Object[]{counterA.get(), hazelcast1.getAtomicLong("counter-a").get()});
    assertThat(countA, is(hazelcast1.getAtomicLong("counter-a").get()));

    assertThat(hazelcast2.getAtomicLong("counter-a").get(), is(hazelcast1.getAtomicLong("counter-a").get()));

    final Member member2 = hazelcast2.getCluster().getLocalMember();
    LOG.log(INFO, member2.getAttributes().entrySet().stream()
            .map(entry -> entry.getKey() + " -> " + entry.getValue() + " : " + entry.getValue().getClass().getName())
            .collect(Collectors.joining("\n", "member 2 attributes\n", "\n"))
    );

    assertThat(member2.getStringAttribute("string"), is("hello"));
    assertThat(member2.getIntAttribute("int"), is(1));
    assertThat(member2.getBooleanAttribute("bool"), is(true));
}

@Override
public void updateJpaAddresses(ConfigurationDao dao) {
    StringBuilder sb = new StringBuilder();
    String delim = "";
    for (Member m : hazelcast.getCluster().getMembers()) {
        sb.append(delim).append(m.getAddress().getHost());
        delim = ";";
    }
    if (Strings.isEmpty(delim)) {
        sb.append("localhost");
    }
    try {
        dao.updateClusterAddresses(sb.toString());
    } catch (UnknownHostException e) {
        log.error("Uexpected exception while updating JPA addresses", e);
        throw new WicketRuntimeException(e);
    }
}

public TabularData getUsageStatistics(Callable<TabularData> statsTask, StatsAggregator aggregator) {
    logger.trace("getUsageStatistics.enter; going to collect stats for schema: {}", schemaName);

    int cnt = 0;
    TabularData result = null;
    Map<Member, Future<TabularData>> futures = execService.submitToAllMembers(statsTask);
    for (Map.Entry<Member, Future<TabularData>> entry: futures.entrySet()) {
        try {
            TabularData stats = entry.getValue().get();
            logger.trace("getUsageStatistics; got stats: {}, from member {}", stats, entry.getKey());
            result = JMXUtils.aggregateStats(stats, result, aggregator);
            logger.trace("getUsageStatistics; got aggregated result: {}", result);
            cnt++;
        } catch (InterruptedException | ExecutionException ex) {
            logger.error("getUsageStatistics.error: " + ex.getMessage(), ex);
        }
    }
    logger.trace("getUsageStatistics.exit; got stats from {} nodes", cnt);
    return result;
}

public LocalNodeConfig find(Member member) throws Exception {
    InetAddress inetAddress = member.getSocketAddress().getAddress();

    String memberIp = inetAddress.getHostAddress();
    String fullHostName = inetAddress.getCanonicalHostName();

    int hazelcastPort = member.getSocketAddress().getPort();

    Set<HazelcastNode> matches = new HashSet<>();
    matches.add(new HazelcastNode(memberIp, hazelcastPort));
    matches.add(new HazelcastNode(fullHostName, hazelcastPort));

    for (HazelcastNode node : nodes.keySet()) {
        if (matches.contains(node)) {
            return nodes.get(node);
        }

    }

    throw new Exception("Member with memberIp <" + memberIp + "> and fullHostName <" + fullHostName + "> and hazelcast port <" + hazelcastPort
                    + "> not found in cluster membership.  Correctly register the machine with server address or ip that all machines can resolve");

}

/**
 * Called on older cluster node when member is removed
 *
 * @param currentMembers
 *            - current cluster members
 * @param memberRemoved
 *            - member that is being removed
 */
public void handleServerRemoved(Set<Member> currentMembers, Member memberRemoved) {
    indexLock.writeLock().lock();
    try {
        Set<Integer> segmentsToRedist = memberToSegmentMap.remove(memberRemoved);
        if (segmentsToRedist != null) {
            Member first = currentMembers.iterator().next();
            memberToSegmentMap.get(first).addAll(segmentsToRedist);
        }

        forceBalance(currentMembers);
    }
    finally {
        indexLock.writeLock().unlock();
    }

}

public void doClusterTask(final ClusterTask task) {
    if (cluster == null) { return; }
    Set<Member> members = new HashSet<Member>();
    Member current = cluster.getLocalMember();
    for(Member member : cluster.getMembers()) {
        if (!member.getUuid().equals(current.getUuid())) {
            members.add(member);
        }
    }
    if (members.size() > 0) {
     // Asynchronously execute the task on the other cluster members
        logger.debug("Executing asynchronous MultiTask: " + task.getClass().getName());
     hazelcast.getExecutorService(HAZELCAST_EXECUTOR_SERVICE_NAME).submitToMembers(
            new CallableTask<Object>(task), members);
    } else {
        logger.warn("No cluster members selected for cluster task " + task.getClass().getName());
    }
}

@ManagedAttribute(description="Returns active schema nodes")
public String[] getActiveNodes() {
    if (schemaInstance == null) {
        return new String[0];
    }
    java.util.Collection<Member> members;
    if (schemaInstance instanceof HazelcastClientInstanceImpl) {
        members = ((HazelcastClientInstanceImpl) schemaInstance).getClientClusterService().getMemberList();
    } else {
        members = schemaInstance.getCluster().getMembers();
    }
    String[] result = new String[members.size()];
    int idx = 0;
    for (Member member: members) {
        result[idx++] = member.getSocketAddress().toString();
    }
    return result;
}

@ManagedAttribute(description="Returns HealthManagement statistics, per node")
public TabularData getHealthStatistics() {
    logger.trace("getHealthStatistics.enter;");
    int cnt = 0;
    TabularData result = null;
    Callable<CompositeData> task = new SchemaHealthAggregator();
    Map<Member, Future<CompositeData>> futures = execService.submitToAllMembers(task);

    for (Map.Entry<Member, Future<CompositeData>> entry: futures.entrySet()) {
        try {
            CompositeData counters = entry.getValue().get();
            logger.trace("getHealthStatistics; got counters: {}, from member {}", counters, entry.getKey());
               result = compositeToTabular("Health", "Desc", "Member", result, counters);
            logger.trace("getHealthStatistics; got aggregated result: {}", result);
            cnt++;
        } catch (InterruptedException | ExecutionException | OpenDataException ex) {
            logger.error("getHealthStatistics.error: " + ex.getMessage(), ex);
        }
    }
    logger.trace("getHealthStatistics.exit; got stats from {} nodes", cnt);
    return result;
}

static <T> ServiceCommandResult<T>[] submit(HazelcastInstance instance, ServiceCommand<T> command) {
  IExecutorService exService = instance.getExecutorService(HAZELCAST_EXECUTOR_NAME) ;
  ServiceCommandWrapper<T> wrapper = new ServiceCommandWrapper<T>(command) ;
  Map<Member, Future<T>>  futures = exService.submitToAllMembers(wrapper) ;
  ServiceCommandResult<T>[] results = new ServiceCommandResult[futures.size()] ;
  long ctime = System.currentTimeMillis() ;
  long waitTime = command.getTimeout() ;
  Iterator<Map.Entry<Member, Future<T>>> i = futures.entrySet().iterator() ;
  int idx  = 0 ;
  while(i.hasNext()) {
    Map.Entry<Member, Future<T>> entry = i.next() ;
    Member hzmember = entry.getKey() ;
    Future<T> future = entry.getValue() ;
    results[idx] = new ServiceCommandResult<T>() ;
    results[idx].setFromMember(new ClusterMemberImpl(hzmember)) ;
    try {
      T ret = future.get(waitTime, TimeUnit.MILLISECONDS) ;
      results[idx].setResult(ret);
    } catch (InterruptedException | ExecutionException | TimeoutException error) {
      results[idx].setError(error) ;
    }
    waitTime = command.getTimeout() - (System.currentTimeMillis() - ctime) ;
    idx++ ;
  }
  return results ;
}

protected boolean onDisconnect(Connection oldConnection) {
    synchronized (callMap) {
        boolean shouldExecuteOnDisconnect = client.getConnectionManager().shouldExecuteOnDisconnect(oldConnection);
        if (!shouldExecuteOnDisconnect) {
            return false;
        }
        Member leftMember = oldConnection.getMember();
        Collection<Call> calls = callMap.values();
        for (Call call : calls) {
            if (call == RECONNECT_CALL) continue;
            Call removed = callMap.remove(call.getId());
            if (removed != null) {
                if (!client.getOutRunnable().queue.contains(removed)) {
                    logger.log(Level.FINE, Thread.currentThread() + ": Calling on disconnect " + leftMember);
                    removed.onDisconnect(leftMember);
                }
            }
        }
        return true;
    }
}

protected int denitSchemaInCluster(Schema schema) {

        logger.trace("denitSchemaInCluster.enter; schema: {}", schema);
        SchemaDenitiator denit = new SchemaDenitiator(schema.getName());

        int cnt = 0; 
        Map<Member, Future<Boolean>> result = execService.submitToAllMembers(denit);
        for (Map.Entry<Member, Future<Boolean>> entry: result.entrySet()) {
            try {
                Boolean ok = entry.getValue().get();
                if (ok) cnt++;
                logger.debug("denitSchemaInCluster; Schema {}de-initialized on node {}", ok ? "" : "NOT ", entry.getKey());
            } catch (InterruptedException | ExecutionException ex) {
                logger.error("denitSchemaInCluster.error; ", ex);
            }
        }
        int rcnt = result.size() - cnt;
        logger.info("denitSchemaInCluster.exit; schema {} de-initialized on {} nodes; returning: {}", 
                schema, cnt, rcnt);
        return rcnt;
    }

public void handleServerAdded(Set<Member> currentMembers, Member memberAdded, boolean master) throws Exception {
    globalLock.writeLock().lock();
    try {
        if (master) {
            // make sure we can resolve it before transferring segments
            Nodes nodes = clusterHelper.getNodes();
            @SuppressWarnings("unused") LocalNodeConfig localNodeConfig = nodes.find(memberAdded);

            handleServerAdded(currentMembers, memberAdded);
        }

        internalClient.addMember(memberAdded);
    }
    finally {
        globalLock.writeLock().unlock();
    }

}

public void readData(DataInput in) throws IOException {
    int memberStatsCount = in.readInt();
    for (int i = 0; i < memberStatsCount; i++) {
        Member member = new MemberImpl();
        member.readData(in);
        int partitionCount = in.readInt();
        int[] partitions = new int[partitionCount];
        for (int a = 0; a < partitionCount; a++) {
            partitions[a] = in.readInt();
        }
        memberPartitions.put(member, partitions);
    }
    int nameCount = in.readInt();
    for (int i = 0; i < nameCount; i++) {
        instanceNames.add(in.readUTF());
    }
}

@Test
public void server_dd() throws Exception {
    final HazelcastInstance hz1 = Hazelcast.newHazelcastInstance();
    final HazelcastInstance hz2 = Hazelcast.newHazelcastInstance();
    final HazelcastInstance hz3 = Hazelcast.newHazelcastInstance();

    for(int i=0; i<1000; i++) {
        server0.getAtomicLong(Integer.toString(i)).incrementAndGet();
    }
    Map<String, Integer> map = server0.getMap("map");
    for (int k = 0; k < 5; k++)
        map.put(UUID.randomUUID().toString(), 1);
    Map<Member, Future<Integer>> executor = server0.getExecutorService("executor").submitToAllMembers(new SumTask());
    for(Map.Entry<Member, Future<Integer>> member : executor.entrySet()) {
        member.getValue().get();
    }

}

public void openConnections(Set<Member> members) throws Exception {

        globalLock.writeLock().lock();
        try {
            Member self = hazelcastManager.getSelf();
            for (Member m : members) {
                if (!self.equals(m)) {
                    internalClient.addMember(m);
                }
            }
        }
        finally {
            globalLock.writeLock().unlock();
        }

    }

@Override
public synchronized void memberAdded(MembershipEvent membershipEvent) {
  if (!active) {
    return;
  }
  try {
    multimaps.forEach(HazelcastAsyncMultiMap::clearCache);
    if (nodeListener != null) {
      Member member = membershipEvent.getMember();
      members.add(member);
      nodeListener.nodeAdded(member.getUuid());
    }
  } catch (Throwable t) {
    log.error("Failed to handle memberAdded", t);
  }
}

public ClusterListener(Cluster cluster) {

    this.cluster = cluster;
    for (Member member : cluster.getMembers()) {
        clusterNodesInfo.put(member.getUuid(), 
                new HazelcastClusterNodeInfo(member, cluster.getClusterTime()));
    }

    C2SCache = CacheFactory.createCache(RoutingTableImpl.C2S_CACHE_NAME);
    anonymousC2SCache = CacheFactory.createCache(RoutingTableImpl.ANONYMOUS_C2S_CACHE_NAME);
    S2SCache = CacheFactory.createCache(RoutingTableImpl.S2S_CACHE_NAME);
    componentsCache = CacheFactory.createCache(RoutingTableImpl.COMPONENT_CACHE_NAME);

    sessionInfoCache = CacheFactory.createCache(SessionManager.C2S_INFO_CACHE_NAME);
    componentSessionsCache = CacheFactory.createCache(SessionManager.COMPONENT_SESSION_CACHE_NAME);
    multiplexerSessionsCache = CacheFactory.createCache(SessionManager.CM_CACHE_NAME);
    incomingServerSessionsCache = CacheFactory.createCache(SessionManager.ISS_CACHE_NAME);

    directedPresencesCache = CacheFactory.createCache(PresenceUpdateHandler.PRESENCE_CACHE_NAME);

    joinCluster();
}

private void returnInternalBlockingConnection(Member m, InternalRpcConnection rpcConnection, boolean valid) {
    InternalRpcConnectionPool connectionPool = internalConnectionPoolMap.get(m);
    if (connectionPool != null) {
        try {
            if (valid) {
                connectionPool.returnObject(rpcConnection);
            }
            else {
                connectionPool.invalidateObject(rpcConnection);
            }
        }
        catch (Exception e) {
            log.error("Failed to return blocking connection to member <" + m + "> pool: ", e);
        }
    }
    else {
        log.error("Failed to return blocking connection to member <" + m + "> pool. Pool does not exist.");
        log.error("Current pool members <" + internalConnectionPoolMap.keySet() + ">");
        if (rpcConnection != null) {
            rpcConnection.close();
        }
    }
}

static <T> ServiceCommandResult<T> submit(HazelcastInstance instance, ServiceCommand<T> command, ClusterMember member) {
  IExecutorService exService = instance.getExecutorService(HAZELCAST_EXECUTOR_NAME) ;
  HazelcastMemberSelector selector = new HazelcastMemberSelector(instance) ;
  Member hzmember = selector.selectMember(member) ;
  ServiceCommandWrapper<T> wrapper = new ServiceCommandWrapper<T>(command) ;
  Future<T> future = exService.submitToMember(wrapper, hzmember) ;
  ServiceCommandResult<T> result = new ServiceCommandResult<T>() ;
  result.setFromMember(member);
  try {
    T ret = future.get(command.getTimeout(), TimeUnit.MILLISECONDS) ;
    result.setResult(ret);
    return result ;
  } catch (InterruptedException | ExecutionException | TimeoutException error) {
    result.setError(error) ;
    return result ;
  }
}

public Member addMember(boolean checkServiceThread, MemberImpl member) {
    if (checkServiceThread) {
        checkServiceThread();
    }
    logger.log(Level.FINEST, "ClusterManager adding " + member);
    if (lsMembers.contains(member)) {
        for (MemberImpl m : lsMembers) {
            if (m.equals(member)) {
                member = m;
            }
        }
        mapMembers.put(member.getAddress(), member);
    } else {
        lsMembers.add(member);
        mapMembers.put(member.getAddress(), member);
    }
    return member;
}

/**
 * {@inheritDoc}
 */
@Override
public List<NodeDetail> getAllNodeDetails() throws AndesException {
    List<NodeDetail> nodeDetails = new ArrayList<>();

    CoordinatorInformation coordinatorDetails = getCoordinatorDetails();
    InetSocketAddress coordinatorSocketAddress = new InetSocketAddress(coordinatorDetails.getHostname(),
            Integer.parseInt(coordinatorDetails.getPort()));

    for (Member member : hazelcastInstance.getCluster().getMembers()) {
        InetSocketAddress nodeSocketAddress = member.getSocketAddress();
        String nodeId = configurableClusterAgent.getIdOfNode(member);
        boolean isCoordinator = nodeSocketAddress.equals(coordinatorSocketAddress);

        nodeDetails.add(new NodeDetail(nodeId, nodeSocketAddress, isCoordinator));
    }

    return nodeDetails;
}