You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ap...@apache.org on 2023/12/08 12:17:04 UTC
(ignite-3) branch main updated: IGNITE-21043 Introduce a method to obtain a list from CompletableFuture#allOf (#2936)
This is an automated email from the ASF dual-hosted git repository.
apolovtsev pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 23238d8222 IGNITE-21043 Introduce a method to obtain a list from CompletableFuture#allOf (#2936)
23238d8222 is described below
commit 23238d822205d0c8f3c44240dba78d3e5f0f4d44
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Fri Dec 8 14:16:58 2023 +0200
IGNITE-21043 Introduce a method to obtain a list from CompletableFuture#allOf (#2936)
---
.../deployunit/DeployMessagingService.java | 44 +++++++--------
.../internal/compute/loader/JobContextManager.java | 63 ++++++++++------------
.../ignite/internal/util/CompletableFutures.java | 24 +++++++++
.../internal/util/CompletableFuturesTest.java | 28 ++++++++++
.../runner/app/AbstractSchemaChangeTest.java | 12 ++---
.../exec/ExecutionDependencyResolverImpl.java | 3 +-
.../internal/table/distributed/TableManager.java | 5 +-
.../replicator/PartitionReplicaListener.java | 32 ++---------
8 files changed, 116 insertions(+), 95 deletions(-)
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java
index 6ef69eb630..397a8f8907 100644
--- a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.deployunit;
+import static java.util.concurrent.CompletableFuture.allOf;
+
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
@@ -24,13 +26,10 @@ import org.apache.ignite.compute.version.Version;
import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
import org.apache.ignite.internal.deployunit.exception.DeploymentUnitReadException;
import org.apache.ignite.internal.deployunit.message.DeployUnitMessageTypes;
+import org.apache.ignite.internal.deployunit.message.DeploymentUnitFactory;
import org.apache.ignite.internal.deployunit.message.DownloadUnitRequest;
-import org.apache.ignite.internal.deployunit.message.DownloadUnitRequestImpl;
import org.apache.ignite.internal.deployunit.message.DownloadUnitResponse;
-import org.apache.ignite.internal.deployunit.message.DownloadUnitResponseImpl;
import org.apache.ignite.internal.deployunit.message.StopDeployRequest;
-import org.apache.ignite.internal.deployunit.message.StopDeployRequestImpl;
-import org.apache.ignite.internal.deployunit.message.StopDeployResponseImpl;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.network.ChannelType;
@@ -65,6 +64,8 @@ public class DeployMessagingService {
*/
private final DownloadTracker tracker;
+ private final DeploymentUnitFactory messageFactory = new DeploymentUnitFactory();
+
/**
* Constructor.
*
@@ -108,7 +109,7 @@ public class DeployMessagingService {
* @return Downloaded deployment unit content.
*/
CompletableFuture<UnitContent> downloadUnitContent(String id, Version version, List<String> nodes) {
- DownloadUnitRequest request = DownloadUnitRequestImpl.builder()
+ DownloadUnitRequest request = messageFactory.downloadUnitRequest()
.id(id)
.version(version.render())
.build();
@@ -137,34 +138,35 @@ public class DeployMessagingService {
* @param version Deployment unit version.
* @return Future with stop result.
*/
- public CompletableFuture<Void> stopInProgressDeploy(String id, Version version) {
+ CompletableFuture<Void> stopInProgressDeploy(String id, Version version) {
LOG.info("Stop in progress deploy for " + id + ":" + version);
- return CompletableFuture.allOf(cmgManager.logicalTopology()
- .thenApply(topology -> topology.nodes().stream().map(node ->
- clusterService.messagingService()
- .invoke(node,
- DEPLOYMENT_CHANNEL,
- StopDeployRequestImpl
- .builder()
- .id(id)
- .version(version.render())
- .build(),
- Long.MAX_VALUE)
- ).toArray(CompletableFuture[]::new)));
+
+ return cmgManager.logicalTopology()
+ .thenCompose(topology -> {
+ StopDeployRequest request = messageFactory.stopDeployRequest()
+ .id(id)
+ .version(version.render())
+ .build();
+
+ CompletableFuture<?>[] sendFutures = topology.nodes().stream()
+ .map(node -> clusterService.messagingService().invoke(node, DEPLOYMENT_CHANNEL, request, Long.MAX_VALUE))
+ .toArray(CompletableFuture[]::new);
+
+ return allOf(sendFutures);
+ });
}
private void processStopDeployRequest(StopDeployRequest request, String senderConsistentId, long correlationId) {
tracker.cancelIfDownloading(request.id(), Version.parseVersion(request.version()));
clusterService.messagingService()
- .respond(senderConsistentId, StopDeployResponseImpl.builder().build(), correlationId);
-
+ .respond(senderConsistentId, messageFactory.stopDeployResponse().build(), correlationId);
}
private void processDownloadRequest(DownloadUnitRequest request, String senderConsistentId, long correlationId) {
deployerService.getUnitContent(request.id(), Version.parseVersion(request.version()))
.thenApply(content -> clusterService.messagingService()
.respond(senderConsistentId,
- DownloadUnitResponseImpl.builder().unitContent(content).build(),
+ messageFactory.downloadUnitResponse().unitContent(content).build(),
correlationId)
);
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobContextManager.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobContextManager.java
index d159751155..012c1c9c14 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobContextManager.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobContextManager.java
@@ -21,13 +21,10 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-import java.lang.reflect.Array;
-import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.stream.Collectors;
-import java.util.stream.IntStream;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.version.Version;
import org.apache.ignite.internal.deployunit.DeploymentStatus;
@@ -39,6 +36,7 @@ import org.apache.ignite.internal.deployunit.exception.DeploymentUnitUnavailable
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.internal.util.RefCountedObjectPool;
import org.apache.ignite.lang.ErrorGroups.Compute;
@@ -100,8 +98,8 @@ public class JobContextManager {
}
/**
- * Creates a class loader for the given deployment units. The units will be acquired.
- * The class loader will be closed when it is not used by any other job.
+ * Creates a class loader for the given deployment units. The units will be acquired. The class loader will be closed when it is not
+ * used by any other job.
*/
private JobClassLoader createClassLoader(List<DeploymentUnit> units) {
List<DisposableDeploymentUnit> disposableDeploymentUnits = units.stream()
@@ -111,8 +109,8 @@ public class JobContextManager {
}
/**
- * Releases a class loader. If the class loader is not used by any other job,
- * it will be closed and the deployment units will be released.
+ * Releases a class loader. If the class loader is not used by any other job, it will be closed and the deployment units will be
+ * released.
*/
private void releaseClassLoader(JobContext jobContext) {
List<DeploymentUnit> units = jobContext.classLoader().units().stream()
@@ -127,11 +125,9 @@ public class JobContextManager {
* Check if the deployment units are deployed to the cluster. Only deployed units can be used in compute.
*/
private CompletableFuture<Void> checkUnitStatuses(List<DeploymentUnit> units) {
- return mapList(units, Void.class, this::checkUnitStatus)
- .thenApply(v -> null);
+ return mapList(units, this::checkUnitStatus, CompletableFuture::allOf);
}
-
private CompletableFuture<Void> checkUnitStatus(DeploymentUnit unit) {
return deployment.clusterStatusAsync(unit.name(), unit.version())
.thenCompose(clusterStatus -> {
@@ -151,24 +147,23 @@ public class JobContextManager {
});
}
-
private CompletableFuture<List<DeploymentUnit>> normalizeVersions(List<DeploymentUnit> units) {
- return mapList(units, DeploymentUnit.class, this::normalizeVersion);
+ return mapList(units, this::normalizeVersion, CompletableFutures::allOf);
}
private CompletableFuture<Void> onDemandDeploy(List<DeploymentUnit> units) {
- return mapList(units, Void.class, it -> {
- return deployment.onDemandDeploy(it.name(), it.version())
- .thenCompose(result -> {
- if (result) {
- return nullCompletedFuture();
- } else {
- return failedFuture(
- new IgniteInternalException(Compute.CLASS_LOADER_ERR, "Failed to deploy on demand unit: " + it.render())
- );
- }
- });
- }).thenApply(ignored -> null);
+ return mapList(
+ units,
+ unit -> deployment.onDemandDeploy(unit.name(), unit.version())
+ .thenAccept(result -> {
+ if (!result) {
+ throw new IgniteInternalException(
+ Compute.CLASS_LOADER_ERR, "Failed to deploy on demand unit: " + unit.render()
+ );
+ }
+ }),
+ CompletableFuture::allOf
+ );
}
private CompletableFuture<DeploymentUnit> normalizeVersion(DeploymentUnit unit) {
@@ -180,17 +175,17 @@ public class JobContextManager {
}
}
- private static <I, O> CompletableFuture<List<O>> mapList(
+ private static <I, O, R> CompletableFuture<R> mapList(
List<I> list,
- Class<O> outputClass,
- Function<I, CompletableFuture<O>> mapper
+ Function<I, CompletableFuture<O>> mapper,
+ Function<CompletableFuture<O>[], CompletableFuture<R>> collector
) {
- O[] accumulator = (O[]) Array.newInstance(outputClass, list.size());
- CompletableFuture<Void>[] futures = IntStream.range(0, list.size())
- .mapToObj(id -> mapper.apply(list.get(id))
- .thenAccept(result -> accumulator[id] = result))
- .<CompletableFuture<Void>>toArray(CompletableFuture[]::new);
- return CompletableFuture.allOf(futures)
- .thenApply(ignored -> Arrays.asList(accumulator));
+ CompletableFuture<O>[] futures = new CompletableFuture[list.size()];
+
+ for (int i = 0; i < list.size(); i++) {
+ futures[i] = mapper.apply(list.get(i));
+ }
+
+ return collector.apply(futures);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/CompletableFutures.java b/modules/core/src/main/java/org/apache/ignite/internal/util/CompletableFutures.java
index 874cbd837d..da9566a0ce 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/CompletableFutures.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/CompletableFutures.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.util;
import static java.util.concurrent.CompletableFuture.completedFuture;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -82,4 +83,27 @@ public class CompletableFutures {
public static <K, V> CompletableFuture<Map<K, V>> emptyMapCompletedFuture() {
return (CompletableFuture<Map<K, V>>) (CompletableFuture<?>) EMPTY_MAP_COMPLETED_FUTURE;
}
+
+ /**
+ * Returns a future that is completed when all provided futures complete (the behavior is identical to {@link CompletableFuture#allOf}).
+ *
+ * <p>If the future completes successfully, it aggregates the results of the source futures into a {@code List}.
+ *
+ * @param cfs Source futures.
+ * @param <T> Type of the result of the source futures.
+ * @return Future that completes with a list of results from the source futures.
+ */
+ @SafeVarargs
+ public static <T> CompletableFuture<List<T>> allOf(CompletableFuture<T>... cfs) {
+ return CompletableFuture.allOf(cfs)
+ .thenApply(v -> {
+ var result = new ArrayList<T>(cfs.length);
+
+ for (CompletableFuture<T> future : cfs) {
+ result.add(future.join());
+ }
+
+ return result;
+ });
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/CompletableFuturesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/CompletableFuturesTest.java
index dccf47b644..63b45c5000 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/CompletableFuturesTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/CompletableFuturesTest.java
@@ -17,6 +17,10 @@
package org.apache.ignite.internal.util;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.will;
import static org.apache.ignite.internal.util.CompletableFutures.booleanCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.emptyCollectionCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.emptyListCompletedFuture;
@@ -25,6 +29,8 @@ import static org.apache.ignite.internal.util.CompletableFutures.emptySetComplet
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -114,4 +120,26 @@ public class CompletableFuturesTest {
assertThrows(UnsupportedOperationException.class, () -> future.join().put(1L, 2));
}
+
+ @Test
+ void testAllOfSuccessFuture() {
+ CompletableFuture<List<Integer>> future = CompletableFutures.allOf(
+ nullCompletedFuture(),
+ completedFuture(1),
+ completedFuture(42)
+ );
+
+ assertThat(future, will(contains(null, 1, 42)));
+ }
+
+ @Test
+ void testAllFailedFuture() {
+ CompletableFuture<List<Integer>> future = CompletableFutures.allOf(
+ nullCompletedFuture(),
+ failedFuture(new RuntimeException("test error")),
+ completedFuture(42)
+ );
+
+ assertThat(future, willThrow(RuntimeException.class, "test error"));
+ }
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/AbstractSchemaChangeTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/AbstractSchemaChangeTest.java
index 49170e4c49..a7b969203a 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/AbstractSchemaChangeTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/AbstractSchemaChangeTest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.runner.app;
import static java.util.stream.Collectors.toList;
-import static java.util.stream.Collectors.toUnmodifiableList;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -35,6 +34,7 @@ import org.apache.ignite.InitParameters;
import org.apache.ignite.internal.IgniteIntegrationTest;
import org.apache.ignite.internal.testframework.TestIgnitionManager;
import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.sql.Session;
@@ -128,9 +128,9 @@ abstract class AbstractSchemaChangeTest extends IgniteIntegrationTest {
* Returns grid nodes.
*/
protected List<Ignite> startGrid() {
- List<CompletableFuture<Ignite>> futures = nodesBootstrapCfg.entrySet().stream()
+ CompletableFuture<Ignite>[] futures = nodesBootstrapCfg.entrySet().stream()
.map(e -> TestIgnitionManager.start(e.getKey(), e.getValue(), workDir.resolve(e.getKey())))
- .collect(toList());
+ .toArray(CompletableFuture[]::new);
String metaStorageNode = nodesBootstrapCfg.keySet().iterator().next();
@@ -142,11 +142,7 @@ abstract class AbstractSchemaChangeTest extends IgniteIntegrationTest {
TestIgnitionManager.init(initParameters);
- await(CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])));
-
- return futures.stream()
- .map(CompletableFuture::join)
- .collect(toUnmodifiableList());
+ return await(CompletableFutures.allOf(futures));
}
/**
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
index 3297117dc9..804e073f89 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
@@ -145,8 +145,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
.thenApply(r -> {
Map<Integer, ExecutableTable> map = tableMap.entrySet()
.stream()
- .map(e -> Map.entry(e.getKey(), e.getValue().join()))
- .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+ .collect(Collectors.toMap(Entry::getKey, e -> e.getValue().join()));
return new ResolvedDependencies(map, dataSources);
});
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index b392464667..b46d5fca24 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -1344,12 +1344,11 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return emptyListCompletedFuture();
}
- CompletableFuture<TableImpl>[] tableImplFutures = tableDescriptors.stream()
+ CompletableFuture<Table>[] tableImplFutures = tableDescriptors.stream()
.map(tableDescriptor -> tableAsyncInternalBusy(tableDescriptor.id()))
.toArray(CompletableFuture[]::new);
- return allOf(tableImplFutures)
- .thenApply(unused1 -> Stream.of(tableImplFutures).map(CompletableFuture::join).collect(toList()));
+ return CompletableFutures.allOf(tableImplFutures);
}), ioExecutor);
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index bf3541af87..0628f26bd6 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -952,17 +952,7 @@ public class PartitionReplicaListener implements ReplicaListener {
resolutionFuts[i] = resolveRowByPkForReadOnly(primaryKeys.get(i), readTimestamp);
}
- return allOf(resolutionFuts).thenApply(unused1 -> {
- var result = new ArrayList<BinaryRow>(resolutionFuts.length);
-
- for (CompletableFuture<BinaryRow> resolutionFut : resolutionFuts) {
- BinaryRow resolvedReadResult = resolutionFut.join();
-
- result.add(resolvedReadResult);
- }
-
- return result;
- });
+ return CompletableFutures.allOf(resolutionFuts);
});
}
@@ -2090,25 +2080,13 @@ public class PartitionReplicaListener implements ReplicaListener {
format("Unknown single request [actionType={}]", request.requestType()));
}
- var resolutionFuts = new ArrayList<CompletableFuture<BinaryRow>>(primaryKeys.size());
-
- for (BinaryTuple primaryKey : primaryKeys) {
- CompletableFuture<BinaryRow> fut = resolveRowByPkForReadOnly(primaryKey, readTimestamp);
+ CompletableFuture<BinaryRow>[] resolutionFuts = new CompletableFuture[primaryKeys.size()];
- resolutionFuts.add(fut);
+ for (int i = 0; i < primaryKeys.size(); i++) {
+ resolutionFuts[i] = resolveRowByPkForReadOnly(primaryKeys.get(i), readTimestamp);
}
- return allOf(resolutionFuts.toArray(new CompletableFuture[0])).thenApply(unused1 -> {
- var result = new ArrayList<BinaryRow>(resolutionFuts.size());
-
- for (CompletableFuture<BinaryRow> resolutionFut : resolutionFuts) {
- BinaryRow resolvedReadResult = resolutionFut.join();
-
- result.add(resolvedReadResult);
- }
-
- return result;
- });
+ return CompletableFutures.allOf(resolutionFuts);
}
/**