You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by su...@apache.org on 2022/12/26 09:36:39 UTC
[shardingsphere] branch master updated: Add ExecuteEngine.trigger (#23103)
This is an automated email from the ASF dual-hosted git repository.
sunnianjun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new aac4039e5db Add ExecuteEngine.trigger (#23103)
aac4039e5db is described below
commit aac4039e5dba69b34fcae9a5c6d2ed9261151597
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Mon Dec 26 17:36:31 2022 +0800
Add ExecuteEngine.trigger (#23103)
* Add ExecuteEngine.trigger
* Improve inner class code style
---
.../data/pipeline/core/execute/ExecuteEngine.java | 23 +++++++
.../core/task/InventoryIncrementalTasksRunner.java | 77 ++++++++++++----------
.../pipeline/core/execute/ExecuteEngineTest.java | 76 +++++++++++++++++++++
3 files changed, 141 insertions(+), 35 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java
index 00988e62067..b92deee1e48 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java
@@ -22,9 +22,11 @@ import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.data.pipeline.api.executor.LifecycleExecutor;
import org.apache.shardingsphere.infra.executor.kernel.thread.ExecutorThreadFactoryBuilder;
+import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Executor engine.
@@ -78,4 +80,25 @@ public final class ExecuteEngine {
}
}, executorService);
}
+
+ /**
+ * Trigger.
+ *
+ * @param futures futures
+ * @param executeCallback execute callback on all the futures
+ */
+ public static void trigger(final Collection<CompletableFuture<?>> futures, final ExecuteCallback executeCallback) {
+ int futureCount = futures.size();
+ AtomicInteger successCount = new AtomicInteger(0);
+ // TODO call onFailure once
+ for (CompletableFuture<?> each : futures) {
+ each.whenComplete((unused, throwable) -> {
+ if (null != throwable) {
+ executeCallback.onFailure(throwable);
+ } else if (successCount.addAndGet(1) == futureCount) {
+ executeCallback.onSuccess();
+ }
+ });
+ }
+ }
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryIncrementalTasksRunner.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryIncrementalTasksRunner.java
index 10848317cbd..086d4e81ef2 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryIncrementalTasksRunner.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryIncrementalTasksRunner.java
@@ -26,13 +26,14 @@ import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.api.task.PipelineTasksRunner;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineJobAPI;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineJobAPIFactory;
+import org.apache.shardingsphere.data.pipeline.core.execute.ExecuteCallback;
+import org.apache.shardingsphere.data.pipeline.core.execute.ExecuteEngine;
import org.apache.shardingsphere.data.pipeline.core.job.PipelineJobIdUtils;
import org.apache.shardingsphere.data.pipeline.core.job.progress.PipelineJobProgressDetector;
import java.util.Collection;
import java.util.LinkedList;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicInteger;
/**
* Inventory incremental tasks' runner.
@@ -93,26 +94,7 @@ public final class InventoryIncrementalTasksRunner implements PipelineTasksRunne
}
futures.addAll(each.start());
}
- AtomicInteger completedCount = new AtomicInteger(0);
- for (CompletableFuture<?> each : futures) {
- each.whenComplete((o, throwable) -> {
- completedCount.addAndGet(1);
- if (null != throwable) {
- log.error("onFailure, inventory task execute failed.", throwable);
- updateLocalAndRemoteJobItemStatus(JobStatus.EXECUTE_INVENTORY_TASK_FAILURE);
- String jobId = jobItemContext.getJobId();
- jobAPI.persistJobItemErrorMessage(jobId, jobItemContext.getShardingItem(), throwable);
- jobAPI.stop(jobId);
- } else if (completedCount.get() == futures.size()) {
- if (PipelineJobProgressDetector.allInventoryTasksFinished(inventoryTasks)) {
- log.info("onSuccess, all inventory tasks finished.");
- executeIncrementalTask();
- } else {
- log.info("onSuccess, inventory tasks not finished");
- }
- }
- });
- }
+ ExecuteEngine.trigger(futures, new InventoryTaskExecuteCallback());
}
private void updateLocalAndRemoteJobItemStatus(final JobStatus jobStatus) {
@@ -137,20 +119,45 @@ public final class InventoryIncrementalTasksRunner implements PipelineTasksRunne
}
futures.addAll(each.start());
}
- AtomicInteger completedCount = new AtomicInteger(0);
- for (CompletableFuture<?> each : futures) {
- each.whenComplete((o, throwable) -> {
- completedCount.addAndGet(1);
- if (null != throwable) {
- log.error("onFailure, incremental task execute failed.", throwable);
- updateLocalAndRemoteJobItemStatus(JobStatus.EXECUTE_INCREMENTAL_TASK_FAILURE);
- String jobId = jobItemContext.getJobId();
- jobAPI.persistJobItemErrorMessage(jobId, jobItemContext.getShardingItem(), throwable);
- jobAPI.stop(jobId);
- } else if (completedCount.get() == futures.size()) {
- log.info("onSuccess, all incremental tasks finished.");
- }
- });
+ ExecuteEngine.trigger(futures, new IncrementalExecuteCallback());
+ }
+
+ private final class InventoryTaskExecuteCallback implements ExecuteCallback {
+
+ @Override
+ public void onSuccess() {
+ if (PipelineJobProgressDetector.allInventoryTasksFinished(inventoryTasks)) {
+ log.info("onSuccess, all inventory tasks finished.");
+ executeIncrementalTask();
+ } else {
+ log.info("onSuccess, inventory tasks not finished");
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ log.error("onFailure, inventory task execute failed.", throwable);
+ updateLocalAndRemoteJobItemStatus(JobStatus.EXECUTE_INVENTORY_TASK_FAILURE);
+ String jobId = jobItemContext.getJobId();
+ jobAPI.persistJobItemErrorMessage(jobId, jobItemContext.getShardingItem(), throwable);
+ jobAPI.stop(jobId);
+ }
+ }
+
+ private final class IncrementalExecuteCallback implements ExecuteCallback {
+
+ @Override
+ public void onSuccess() {
+ log.info("onSuccess, all incremental tasks finished.");
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ log.error("onFailure, incremental task execute failed.", throwable);
+ updateLocalAndRemoteJobItemStatus(JobStatus.EXECUTE_INCREMENTAL_TASK_FAILURE);
+ String jobId = jobItemContext.getJobId();
+ jobAPI.persistJobItemErrorMessage(jobId, jobItemContext.getShardingItem(), throwable);
+ jobAPI.stop(jobId);
}
}
}
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/execute/ExecuteEngineTest.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/execute/ExecuteEngineTest.java
index 226a330d68f..976b5da7da0 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/execute/ExecuteEngineTest.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/execute/ExecuteEngineTest.java
@@ -17,6 +17,7 @@
package org.apache.shardingsphere.test.it.data.pipeline.core.execute;
+import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.apache.shardingsphere.data.pipeline.api.executor.LifecycleExecutor;
import org.apache.shardingsphere.data.pipeline.core.execute.ExecuteCallback;
@@ -24,10 +25,14 @@ import org.apache.shardingsphere.data.pipeline.core.execute.ExecuteEngine;
import org.junit.Test;
import org.mockito.internal.configuration.plugins.Plugins;
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -77,4 +82,75 @@ public final class ExecuteEngineTest {
executorService.shutdown();
executorService.awaitTermination(30L, TimeUnit.SECONDS);
}
+
+ @Test
+ public void assertTriggerAllFailure() throws InterruptedException {
+ CompletableFuture<?> future1 = CompletableFuture.runAsync(new FixtureRunnable(false));
+ CompletableFuture<?> future2 = CompletableFuture.runAsync(new FixtureRunnable(false));
+ FixtureExecuteCallback executeCallback = new FixtureExecuteCallback(2);
+ ExecuteEngine.trigger(Arrays.asList(future1, future2), executeCallback);
+ executeCallback.latch.await();
+ assertThat(executeCallback.successCount.get(), is(0));
+ assertThat(executeCallback.failureCount.get(), is(2));
+ }
+
+ @Test
+ public void assertTriggerAllSuccess() throws InterruptedException {
+ CompletableFuture<?> future1 = CompletableFuture.runAsync(new FixtureRunnable(true));
+ CompletableFuture<?> future2 = CompletableFuture.runAsync(new FixtureRunnable(true));
+ FixtureExecuteCallback executeCallback = new FixtureExecuteCallback(1);
+ ExecuteEngine.trigger(Arrays.asList(future1, future2), executeCallback);
+ executeCallback.latch.await();
+ assertThat(executeCallback.successCount.get(), is(1));
+ assertThat(executeCallback.failureCount.get(), is(0));
+ }
+
+ @Test
+ public void assertTriggerPartSuccessFailure() throws InterruptedException {
+ CompletableFuture<?> future1 = CompletableFuture.runAsync(new FixtureRunnable(true));
+ CompletableFuture<?> future2 = CompletableFuture.runAsync(new FixtureRunnable(false));
+ FixtureExecuteCallback executeCallback = new FixtureExecuteCallback(1);
+ ExecuteEngine.trigger(Arrays.asList(future1, future2), executeCallback);
+ executeCallback.latch.await();
+ assertThat(executeCallback.successCount.get(), is(0));
+ assertThat(executeCallback.failureCount.get(), is(1));
+ }
+
+ @RequiredArgsConstructor
+ private static final class FixtureRunnable implements Runnable {
+
+ private final boolean success;
+
+ @Override
+ public void run() {
+ if (!success) {
+ throw new RuntimeException("Failure mock");
+ }
+ }
+ }
+
+ private static final class FixtureExecuteCallback implements ExecuteCallback {
+
+ private final CountDownLatch latch;
+
+ private final AtomicInteger successCount = new AtomicInteger(0);
+
+ private final AtomicInteger failureCount = new AtomicInteger(0);
+
+ FixtureExecuteCallback(final int latchCount) {
+ this.latch = new CountDownLatch(latchCount);
+ }
+
+ @Override
+ public void onSuccess() {
+ successCount.addAndGet(1);
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ failureCount.addAndGet(1);
+ latch.countDown();
+ }
+ }
}