test_class.declarator: NODE_COUNT = 10
test_class.modifier: private static final
test_class.original_string: private static final int NODE_COUNT = 10;
test_class.type: int
test_class.var_name: NODE_COUNT
test_class.declarator: ignite
test_class.modifier: private
test_class.original_string: private Ignite ignite;
test_class.type: Ignite
test_class.var_name: ignite
test_class.file: modules/ml/src/test/java/org/apache/ignite/ml/dataset/impl/cache/util/ComputeUtilsTest.java
test_class.identifier: ComputeUtilsTest
test_class.superclass: extends GridCommonAbstractTest
test_case.body: @Test
      public void testAffinityCallWithRetriesNegative() {
          ClusterNode node1 = grid(1).cluster().localNode();
          ClusterNode node2 = grid(2).cluster().localNode();
  
          String firstCacheName = "CACHE_1_" + UUID.randomUUID();
          String secondCacheName = "CACHE_2_" + UUID.randomUUID();
  
          CacheConfiguration<Integer, Integer> cacheConfiguration1 = new CacheConfiguration<>();
          cacheConfiguration1.setName(firstCacheName);
          cacheConfiguration1.setAffinity(new TestAffinityFunction(node1));
          IgniteCache<Integer, Integer> cache1 = ignite.createCache(cacheConfiguration1);
  
          CacheConfiguration<Integer, Integer> cacheConfiguration2 = new CacheConfiguration<>();
          cacheConfiguration2.setName(secondCacheName);
          cacheConfiguration2.setAffinity(new TestAffinityFunction(node2));
          IgniteCache<Integer, Integer> cache2 = ignite.createCache(cacheConfiguration2);
  
          try {
              try {
                  ComputeUtils.affinityCallWithRetries(
                      ignite,
                      Arrays.asList(firstCacheName, secondCacheName),
                      part -> part,
                      0,
                      DeployingContext.unitialized()
                  );
              }
              catch (IllegalStateException expectedException) {
                  return;
              }
  
              fail("Missing IllegalStateException");
          }
          finally {
              cache1.destroy();
              cache2.destroy();
          }
      }
test_case.class_method_signature: ComputeUtilsTest.testAffinityCallWithRetriesNegative()
test_case.full_signature: @Test public void testAffinityCallWithRetriesNegative()
test_case.identifier: testAffinityCallWithRetriesNegative
test_case.invocations: localNode cluster grid localNode cluster grid randomUUID randomUUID setName setAffinity createCache setName setAffinity createCache affinityCallWithRetries asList unitialized fail destroy destroy
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testAffinityCallWithRetriesNegative()
focal_class.declarator: DATA_STORAGE_KEY_TEMPLATE = "part_data_storage_%s"
focal_class.modifier: private static final
focal_class.original_string: private static final String DATA_STORAGE_KEY_TEMPLATE = "part_data_storage_%s";
focal_class.type: String
focal_class.var_name: DATA_STORAGE_KEY_TEMPLATE
focal_class.declarator: ENVIRONMENT_STORAGE_KEY_TEMPLATE = "part_environment_storage_%s"
focal_class.modifier: private static final
focal_class.original_string: private static final String ENVIRONMENT_STORAGE_KEY_TEMPLATE = "part_environment_storage_%s";
focal_class.type: String
focal_class.var_name: ENVIRONMENT_STORAGE_KEY_TEMPLATE
focal_class.file: modules/ml/src/main/java/org/apache/ignite/ml/dataset/impl/cache/util/ComputeUtils.java
focal_class.identifier: ComputeUtils
focal_class.class_method_signature: ComputeUtils.affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.full_signature: public static Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.identifier: affinityCallWithRetries
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.return: Collection<R>
focal_class.signature: Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_class.class_method_signature: ComputeUtils.affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.full_signature: public static Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.identifier: affinityCallWithRetries
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.return: Collection<R>
focal_class.signature: Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, DeployingContext deployingContext)
focal_class.class_method_signature: ComputeUtils.getLearningEnvironment(Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.full_signature: public static LearningEnvironment getLearningEnvironment(Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.identifier: getLearningEnvironment
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.return: LearningEnvironment
focal_class.signature: LearningEnvironment getLearningEnvironment(Ignite ignite,
          UUID datasetId,
          int part,
          LearningEnvironmentBuilder envBuilder)
focal_class.class_method_signature: ComputeUtils.getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.full_signature: public static D getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.identifier: getData
focal_class.modifiers: public static
focal_class.parameters: (
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.return: D
focal_class.signature: D getData(
          Ignite ignite,
          String upstreamCacheName, IgniteBiPredicate<K, V> filter,
          UpstreamTransformerBuilder transformerBuilder,
          String datasetCacheName, UUID datasetId,
          PartitionDataBuilder<K, V, C, D> partDataBuilder,
          LearningEnvironment env,
          boolean isKeepBinary)
focal_class.class_method_signature: ComputeUtils.removeData(Ignite ignite, UUID datasetId)
focal_class.full_signature: public static void removeData(Ignite ignite, UUID datasetId)
focal_class.identifier: removeData
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, UUID datasetId)
focal_class.return: void
focal_class.signature: void removeData(Ignite ignite, UUID datasetId)
focal_class.class_method_signature: ComputeUtils.removeLearningEnv(Ignite ignite, UUID datasetId)
focal_class.full_signature: public static void removeLearningEnv(Ignite ignite, UUID datasetId)
focal_class.identifier: removeLearningEnv
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, UUID datasetId)
focal_class.return: void
focal_class.signature: void removeLearningEnv(Ignite ignite, UUID datasetId)
focal_class.class_method_signature: ComputeUtils.initContext(
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.full_signature: public static void initContext(
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.identifier: initContext
focal_class.modifiers: public static
focal_class.parameters: (
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.return: void
focal_class.signature: void initContext(
          Ignite ignite,
          String upstreamCacheName,
          UpstreamTransformerBuilder transformerBuilder,
          IgniteBiPredicate<K, V> filter,
          String datasetCacheName,
          PartitionContextBuilder<K, V, C> ctxBuilder,
          LearningEnvironmentBuilder envBuilder,
          int retries,
          int interval,
          boolean isKeepBinary,
          DeployingContext deployingContext)
focal_class.class_method_signature: ComputeUtils.getContext(Ignite ignite, String datasetCacheName, int part)
focal_class.full_signature: public static C getContext(Ignite ignite, String datasetCacheName, int part)
focal_class.identifier: getContext
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, String datasetCacheName, int part)
focal_class.return: C
focal_class.signature: C getContext(Ignite ignite, String datasetCacheName, int part)
focal_class.class_method_signature: ComputeUtils.saveContext(Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.full_signature: public static void saveContext(Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.identifier: saveContext
focal_class.modifiers: public static
focal_class.parameters: (Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.return: void
focal_class.signature: void saveContext(Ignite ignite, String datasetCacheName, int part, C ctx)
focal_class.class_method_signature: ComputeUtils.computeCount(
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.full_signature: private static long computeCount(
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.identifier: computeCount
focal_class.modifiers: private static
focal_class.parameters: (
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.return: long
focal_class.signature: long computeCount(
          IgniteCache<K, V> cache,
          ScanQuery<K, V> qry,
          UpstreamTransformer transformer)
focal_class.class_method_signature: ComputeUtils.computeCount(Iterator<?> iter)
focal_class.full_signature: private static long computeCount(Iterator<?> iter)
focal_class.identifier: computeCount
focal_class.modifiers: private static
focal_class.parameters: (Iterator<?> iter)
focal_class.return: long
focal_class.signature: long computeCount(Iterator<?> iter)
focal_method.body: public static <R> Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext) {
          assert !cacheNames.isEmpty();
          assert interval >= 0;
  
          String primaryCache = cacheNames.iterator().next();
  
          Affinity<?> affinity = ignite.affinity(primaryCache);
          int partitions = affinity.partitions();
  
          BitSet completionFlags = new BitSet(partitions);
          Collection<R> results = new ArrayList<>();
  
          for (int t = 0; t <= retries; t++) {
              ClusterGroup clusterGrp = ignite.cluster().forDataNodes(primaryCache);
  
              // Sends jobs.
              Map<Integer, IgniteFuture<R>> futures = new HashMap<>();
              for (int part = 0; part < partitions; part++)
                  if (!completionFlags.get(part)) {
                      final int currPart = part;
  
                      futures.put(
                          currPart,
                          ignite.compute(clusterGrp).affinityCallAsync(
                              cacheNames, currPart,
                              new DeployableCallable<>(deployingContext, part, fun)
                          )
                      );
                  }
  
              // Collects results.
              for (int part : futures.keySet())
                  try {
                      R res = futures.get(part).get();
                      results.add(res);
                      completionFlags.set(part);
                  }
                  catch (IgniteException ignore) {
                  }
  
              if (completionFlags.cardinality() == partitions)
                  return results;
  
              LockSupport.parkNanos(interval * 1_000_000);
          }
  
          throw new IllegalStateException();
      }
focal_method.class_method_signature: ComputeUtils.affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_method.full_signature: public static Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_method.identifier: affinityCallWithRetries
focal_method.invocations: isEmpty next iterator affinity partitions forDataNodes cluster get put affinityCallAsync compute keySet get get add set cardinality parkNanos
focal_method.modifiers: public static
focal_method.parameters: (Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)
focal_method.return: Collection<R>
focal_method.signature: Collection<R> affinityCallWithRetries(Ignite ignite, Collection<String> cacheNames,
          IgniteFunction<Integer, R> fun, int retries, int interval, DeployingContext deployingContext)