import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * Add a distributed step which works in the following way:
 * 1. apply local context and input to local context extractor and keys supplier to get corresponding suppliers;
 * 2. on each node_n
 * 2.1. get context object.
 * 2.2. for each entry_i e located on node_n with key_i from keys stream compute worker((context, entry_i)) and get
 * (cachesUpdates_i, result_i).
 * 2.3. for all i on node_n merge cacheUpdates_i and apply them.
 * 2.4. for all i on node_n, reduce result_i into result_n.
 * 3. get all result_n, reduce them into result and return result.
 *
 * @param <O1> Type of worker output.
 * @param <G> Type of context used by worker.
 * @param workerCtxExtractor Extractor of context for worker.
 * @param worker Function computed on each entry of cache used for training. Second argument is context:
 * common part of data which is independent from key.
 * @param ks Function from chain input and local context to supplier of keys for worker.
 * @param reducer Function used for reducing results of worker.
 * @return Combination of this chain and distributed step specified by given parameters.
 */
default <O1 extends Serializable, G> ComputationsChain<L, K, V, I, O1> thenDistributedForEntries(
        IgniteBiFunction<O, L, IgniteSupplier<G>> workerCtxExtractor,
        IgniteFunction<EntryAndContext<K, V, G>, ResultAndUpdates<O1>> worker,
        IgniteBiFunction<O, L, IgniteSupplier<Stream<GroupTrainerCacheKey<K>>>> ks,
        IgniteFunction<List<O1>, O1> reducer) {
    ComputationsChain<L, K, V, O, O1> nextStep = (input, context) -> {
        L locCtx = context.localContext();
        IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysSupplier = ks.apply(input, locCtx);

        Ignite ignite = context.ignite();
        UUID trainingUUID = context.localContext().trainingUUID();
        String cacheName = context.cache().getName();
        ClusterGroup grp = ignite.cluster().forDataNodes(cacheName);

        // Apply first two arguments locally because it is common for all nodes.
        IgniteSupplier<G> extractor = Functions.curry(workerCtxExtractor).apply(input).apply(locCtx);

        return ignite.compute(grp).execute(new GroupTrainerEntriesProcessorTask<>(trainingUUID, extractor, worker, keysSupplier, reducer, cacheName, ignite), null);
    };
    return then(nextStep);
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * Add a distributed step which works in the following way:
 * 1. apply local context and input to local context extractor and keys supplier to get corresponding suppliers;
 * 2. on each node_n
 * 2.1. get context object.
 * 2.2. for each key_i from keys stream such that key_i located on node_n compute worker((context, entry_i)) and get
 * (cachesUpdates_i, result_i).
 * 2.3. for all i on node_n merge cacheUpdates_i and apply them.
 * 2.4. for all i on node_n, reduce result_i into result_n.
 * 3. get all result_n, reduce them into result and return result.
 *
 * @param <O1> Type of worker output.
 * @param <G> Type of context used by worker.
 * @param workerCtxExtractor Extractor of context for worker.
 * @param worker Function computed on each entry of cache used for training. Second argument is context:
 * common part of data which is independent from key.
 * @param keysSupplier Function from chain input and local context to supplier of keys for worker.
 * @param reducer Function used for reducing results of worker.
 * @return Combination of this chain and distributed step specified by given parameters.
 */
default <O1 extends Serializable, G> ComputationsChain<L, K, V, I, O1> thenDistributedForKeys(
        IgniteBiFunction<O, L, IgniteSupplier<G>> workerCtxExtractor,
        IgniteFunction<KeyAndContext<K, G>, ResultAndUpdates<O1>> worker,
        IgniteBiFunction<O, L, IgniteSupplier<Stream<GroupTrainerCacheKey<K>>>> keysSupplier,
        IgniteFunction<List<O1>, O1> reducer) {
    ComputationsChain<L, K, V, O, O1> nextStep = (input, context) -> {
        L locCtx = context.localContext();
        IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> ks = keysSupplier.apply(input, locCtx);

        Ignite ignite = context.ignite();
        UUID trainingUUID = context.localContext().trainingUUID();
        String cacheName = context.cache().getName();
        ClusterGroup grp = ignite.cluster().forDataNodes(cacheName);

        // Apply first argument locally because it is common for all nodes.
        IgniteSupplier<G> extractor = Functions.curry(workerCtxExtractor).apply(input).apply(locCtx);

        return ignite.compute(grp).execute(new GroupTrainerKeysProcessorTask<>(trainingUUID, extractor, worker, ks, reducer, cacheName, ignite), null);
    };
    return then(nextStep);
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testNewNodes() throws Exception {
    ClusterGroup youngest = ignite.cluster().forYoungest();
    ClusterGroup oldest = ignite.cluster().forOldest();

    ClusterNode old = oldest.node();
    ClusterNode last = youngest.node();

    assertNotNull(last);

    try (Ignite g = startGrid(NODES_CNT)) {
        ClusterNode n = g.cluster().localNode();

        ClusterNode latest = youngest.node();

        assertNotNull(latest);
        assertEquals(latest.id(), n.id());
        assertEquals(oldest.node(), old);
    }
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * Add a distributed step which works in the following way:
 * 1. apply local context and input to local context extractor and keys supplier to get corresponding suppliers;
 * 2. on each node_n
 * 2.1. get context object.
 * 2.2. for each entry_i e located on node_n with key_i from keys stream compute worker((context, entry_i)) and get
 * (cachesUpdates_i, result_i).
 * 2.3. for all i on node_n merge cacheUpdates_i and apply them.
 * 2.4. for all i on node_n, reduce result_i into result_n.
 * 3. get all result_n, reduce them into result and return result.
 *
 * @param <O1> Type of worker output.
 * @param <G> Type of context used by worker.
 * @param workerCtxExtractor Extractor of context for worker.
 * @param worker Function computed on each entry of cache used for training. Second argument is context:
 * common part of data which is independent from key.
 * @param ks Function from chain input and local context to supplier of keys for worker.
 * @param reducer Function used for reducing results of worker.
 * @return Combination of this chain and distributed step specified by given parameters.
 */
default <O1 extends Serializable, G> ComputationsChain<L, K, V, I, O1> thenDistributedForEntries(
    IgniteBiFunction<O, L, IgniteSupplier<G>> workerCtxExtractor,
    IgniteFunction<EntryAndContext<K, V, G>, ResultAndUpdates<O1>> worker,
    IgniteBiFunction<O, L, IgniteSupplier<Stream<GroupTrainerCacheKey<K>>>> ks,
    IgniteFunction<List<O1>, O1> reducer) {
    ComputationsChain<L, K, V, O, O1> nextStep = (input, context) -> {
        L locCtx = context.localContext();
        IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysSupplier = ks.apply(input, locCtx);

        Ignite ignite = context.ignite();
        UUID trainingUUID = context.localContext().trainingUUID();
        String cacheName = context.cache().getName();
        ClusterGroup grp = ignite.cluster().forDataNodes(cacheName);

        // Apply first two arguments locally because it is common for all nodes.
        IgniteSupplier<G> extractor = Functions.curry(workerCtxExtractor).apply(input).apply(locCtx);

        return ignite.compute(grp).execute(new GroupTrainerEntriesProcessorTask<>(trainingUUID, extractor, worker, keysSupplier, reducer, cacheName, ignite), null);
    };
    return then(nextStep);
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testForOthers() throws Exception {
    ClusterNode node0 = grid(0).localNode();
    ClusterNode node1 = grid(1).localNode();
    ClusterNode node2 = grid(2).localNode();
    ClusterNode node3 = grid(3).localNode();

    ClusterGroup p1 = grid(0).cluster().forOthers(node0);

    assertEquals(3, p1.nodes().size());

    assertEquals(2, p1.forOthers(node1).nodes().size());

    assertEquals(1, p1.forOthers(node1, node2).nodes().size());

    assertEquals(1, grid(0).cluster().forOthers(node1, node2, node3).nodes().size());
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * Starts the background cleanup of old cache entries.
 *
 * @param grid Grid.
 * @param metaCache Meta cache.
 * @param dataCacheName Data cache name.
 * @param currentVersions Current versions.
 */
private void startBackgroundCleanup(Ignite grid, final Cache<CleanupNodeId, UUID> metaCache,
    final String dataCacheName, final Map<String, EntryProcessorResult<Long>> currentVersions) {
    if (cleanupFlags.containsKey(dataCacheName))
        return;  // Current node already performs cleanup.

    if (!trySetGlobalCleanupFlag(grid, metaCache))
        return;

    cleanupFlags.put(dataCacheName, true);

    final ClusterGroup dataNodes = grid.cluster().forDataNodes(dataCacheName);

    IgniteFuture f = grid.compute(dataNodes).broadcastAsync(
        new RemoveOldEntriesRunnable(dataCacheName, currentVersions));

    f.listen(new CleanupCompletionListener(metaCache, dataCacheName));
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/** @throws Exception If failed. */
@SuppressWarnings("deprecation")
public void testProjectionAffinity() throws Exception {
    waitTopologyUpdate();

    Ignite g0 = grid(0);
    Ignite g1 = grid(1);

    ClusterGroup g0Pinned = g0.cluster().forNodeIds(F.asList(g0.cluster().localNode().id()));

    ClusterGroup g01Pinned =
        g1.cluster().forNodeIds(F.asList(g0.cluster().localNode().id(), g1.cluster().localNode().id()));

    for (int i = 0; i < 100; i++)
        assertEquals(g0Pinned.ignite().affinity(DEFAULT_CACHE_NAME).mapKeyToNode(i).id(),
            g01Pinned.ignite().affinity(DEFAULT_CACHE_NAME).mapKeyToNode(i).id());
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testTransformResourceInjection() throws Exception {
    ClusterGroup servers = grid(0).cluster().forServers();

    if(F.isEmpty(servers.nodes()))
        return;

    grid(0).services( grid(0).cluster()).deployNodeSingleton(SERVICE_NAME1, new DummyServiceImpl());

    IgniteCache<String, Integer> cache = jcache();
    Ignite ignite = ignite(0);

    doTransformResourceInjection(ignite, cache, false, false);
    doTransformResourceInjection(ignite, cache, true, false);
    doTransformResourceInjection(ignite, cache, true, true);

    if (txEnabled()) {
        doTransformResourceInjectionInTx(ignite, cache, false, false);
        doTransformResourceInjectionInTx(ignite, cache, true, false);
        doTransformResourceInjectionInTx(ignite, cache, true, true);
    }
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testEmptyGroup() throws Exception {
    ClusterGroup emptyGrp = ignite.cluster().forAttribute("nonExistent", "val");

    assertEquals(0, emptyGrp.forOldest().nodes().size());
    assertEquals(0, emptyGrp.forYoungest().nodes().size());
    assertEquals(0, emptyGrp.forAttribute("nonExistent2", "val").nodes().size());
    assertEquals(0, emptyGrp.forCacheNodes("cacheName").nodes().size());
    assertEquals(0, emptyGrp.forClientNodes("cacheName").nodes().size());
    assertEquals(0, emptyGrp.forClients().nodes().size());
    assertEquals(0, emptyGrp.forDaemons().nodes().size());
    assertEquals(0, emptyGrp.forDataNodes("cacheName").nodes().size());
    assertEquals(0, emptyGrp.forRandom().nodes().size());
    assertEquals(0, emptyGrp.forRemotes().nodes().size());
    assertEquals(0, emptyGrp.forServers().nodes().size());
    assertEquals(0, emptyGrp.forHost(ignite.cluster().localNode()).nodes().size());
    assertEquals(0, emptyGrp.forHost("127.0.0.1").nodes().size());
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public ClusterGroup forPredicate(IgnitePredicate<ClusterNode> p) {
    A.notNull(p, "p");

    guard();

    try {
        if (p != null)
            ctx.resource().injectGeneric(p);

        return new ClusterGroupAdapter(ctx, subjId, this.p != null ? F.and(p, this.p) : p);
    }
    catch (IgniteCheckedException e) {
        throw U.convertException(e);
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testApply3() throws Exception {
    testMasterLeaveAwareCallback(2, new CX1<ClusterGroup, IgniteFuture<?>>() {
        @Override public IgniteFuture<?> applyx(ClusterGroup grid) {
            return compute(grid).applyAsync(new TestClosure(),
                Arrays.asList("arg1", "arg2"),
                new IgniteReducer<Void, Object>() {
                    @Override public boolean collect(@Nullable Void aVoid) {
                        return true;
                    }

                    @Override public Object reduce() {
                        return null;
                    }
                });
        }
    });
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * Single server test.
 *
 * @param async Async message send flag.
 * @throws Exception If failed.
 */
private void localServerInternal(boolean async) throws Exception {
    int messages = MSGS;

    Ignite ignite = grid(SERVER_NODE_IDX);

    LATCH = new CountDownLatch(messages);

    ClusterGroup grp = grid(SERVER_NODE_IDX).cluster().forLocal();

    UUID opId = registerListener(grp);

    try {
        for (int i = 0; i < messages; i++)
            sendMessage(ignite, grp, value(i), async);

        assertTrue(LATCH.await(10, TimeUnit.SECONDS));

    }
    finally {
        ignite.message().stopRemoteListen(opId);
    }
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public final ClusterGroup forNodes(Collection<? extends ClusterNode> nodes) {
    A.notEmpty(nodes, "nodes");

    guard();

    try {
        Set<UUID> nodeIds = U.newHashSet(nodes.size());

        for (ClusterNode n : nodes)
            if (contains(n))
                nodeIds.add(n.id());

        return new ClusterGroupAdapter(ctx, subjId, nodeIds);
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public final ClusterGroup forNodeIds(Collection<UUID> ids) {
    A.notEmpty(ids, "ids");

    guard();

    try {
        Set<UUID> nodeIds = U.newHashSet(ids.size());

        for (UUID id : ids) {
            if (contains(id))
                nodeIds.add(id);
        }

        return new ClusterGroupAdapter(ctx, subjId, nodeIds);
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * @param excludeIds Node IDs.
 * @return New cluster group.
 */
private ClusterGroup forOthers(Collection<UUID> excludeIds) {
    assert excludeIds != null;

    if (ids != null) {
        guard();

        try {
            Set<UUID> nodeIds = U.newHashSet(ids.size());

            for (UUID id : ids) {
                if (!excludeIds.contains(id))
                    nodeIds.add(id);
            }

            return new ClusterGroupAdapter(ctx, subjId, nodeIds);
        }
        finally {
            unguard();
        }
    }
    else
        return forPredicate(new OthersFilter(excludeIds));
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
public void startReconciliation(UUID replicaId, long startTransactionId) {
    LOGGER.info("[M] Reconciliation started for replica {} and tx id {}", UUIDFormat.f(replicaId), startTransactionId);
    resubscribeReplicaIfIsOutOfOrder(replicaId);
    if (startTransactionId != LeadContextLoader.NOT_LOADED) {
        kafkaService.seekToTransaction(
            dataRecoveryConfig,
            startTransactionId,
            kafkaFactory,
            replicaService.toGroupId(replicaId));
    }
    ReplicaConfig config = replicaService.getReplicaConfig(replicaId);
    ClusterGroup clusterGroup = clusterGroupService.getReconciliationClusterGroup(ignite, kafkaFactory, config);
    ignite.compute(clusterGroup).broadcast(new StartReconciliationJob(replicaId));
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
public void start() {
    stopAll(); // Stop previous context loading process if running.
    String groupId = LEAD_LOADER_GROUP_ID_PREFIX + leadId;
    initState(groupId);
    ClusterGroup subcluster = clusterGroupService.getLeadContextLoadingClusterGroup(ignite, kafkaFactory,
        dataRecoveryConfig);

    numberOfLoadingJobs = subcluster.nodes().size();
    LOGGER.info("[L] Start lead load with {}", numberOfLoadingJobs);
    stalledLoadingJobs = new HashSet<>(numberOfLoadingJobs);
    ignite.compute(subcluster).broadcast(new StartContextLoadingJob(leadId, groupId));
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
private ClusterGroup getClusterGroupAsSubcollection(Ignite ignite, int size) {
    Collection<ClusterNode> clusterNodes = ignite.cluster().forServers().nodes();

    if (size >= clusterNodes.size()) {
        return ignite.cluster().forServers();
    }
    Iterator<ClusterNode> clusterNodeIter = clusterNodes.iterator();
    List<ClusterNode> groupNodes = new ArrayList<>(size);

    while (groupNodes.size() < size) {
        groupNodes.add(clusterNodeIter.next());
    }
    return ignite.cluster().forNodes(groupNodes);
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
@Before
public void init() {
    MockitoAnnotations.initMocks(this);

    ClusterGroup group = mock(ClusterGroup.class);
    IgniteCompute compute = mock(IgniteCompute.class);
    doReturn(compute).when(ignite).compute();
    doReturn(compute).when(ignite).compute(group);
    doReturn(group).when(clusterGroupService).getLeadContextLoadingClusterGroup(ignite, kafkaFactory, dataRecoveryConfig);

    leadContextLoader = new LeadContextLoader(ignite, kafkaFactory, UUID.randomUUID(), dataRecoveryConfig,
        lastDenseCommitted, clusterGroupService, kafkaService);
}
 

import org.apache.ignite.cluster.ClusterGroup; //导入依赖的package包/类
/**
 * Checks minimum topology size for running a certain example.
 *
 * @param grp Cluster to check size for.
 * @param size Minimum number of nodes required to run a certain example.
 * @return {@code True} if check passed, {@code false} otherwise.
 */
public static boolean checkMinTopologySize(ClusterGroup grp, int size) {
    int prjSize = grp.nodes().size();

    if (prjSize < size) {
        System.err.println(">>> Please start at least " + size + " cluster nodes to run example.");

        return false;
    }

    return true;
}