You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by pa...@apache.org on 2022/05/17 10:32:15 UTC
[shardingsphere] branch master updated: Simplify ExecuteEngine and complete ExecuteEngineTest (#17739)
This is an automated email from the ASF dual-hosted git repository.
panjuan 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 60c847983ce Simplify ExecuteEngine and complete ExecuteEngineTest (#17739)
60c847983ce is described below
commit 60c847983cee70d4b5b58985f6f9be1c45a67c11
Author: 吴伟杰 <wu...@apache.org>
AuthorDate: Tue May 17 18:32:09 2022 +0800
Simplify ExecuteEngine and complete ExecuteEngineTest (#17739)
---
.../data/pipeline/core/execute/ExecuteEngine.java | 20 +--
.../pipeline/core/execute/ExecuteEngineTest.java | 137 +++++++++++++--------
2 files changed, 92 insertions(+), 65 deletions(-)
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java
index fb743a5cbe9..43d441b6873 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngine.java
@@ -63,16 +63,6 @@ public final class ExecuteEngine {
return new ExecuteEngine(Executors.newFixedThreadPool(threadNumber, ExecutorThreadFactoryBuilder.build(threadNameFormat)));
}
- /**
- * Submit a {@code LifecycleExecutor} without callback to execute.
- *
- * @param lifecycleExecutor lifecycle executor
- * @return execute future
- */
- public Future<?> submit(final LifecycleExecutor lifecycleExecutor) {
- return executorService.submit(lifecycleExecutor);
- }
-
/**
* Submit a {@code LifecycleExecutor} with callback {@code ExecuteCallback} to execute.
*
@@ -85,7 +75,8 @@ public final class ExecuteEngine {
if (null == throwable) {
executeCallback.onSuccess();
} else {
- executeCallback.onFailure(throwable);
+ Throwable cause = throwable.getCause();
+ executeCallback.onFailure(null != cause ? cause : throwable);
}
}, executorService);
}
@@ -104,10 +95,11 @@ public final class ExecuteEngine {
futures[i++] = CompletableFuture.runAsync(each, executorService);
}
return CompletableFuture.allOf(futures).whenCompleteAsync((unused, throwable) -> {
- if (null != throwable) {
- executeCallback.onFailure(throwable);
- } else {
+ if (null == throwable) {
executeCallback.onSuccess();
+ } else {
+ Throwable cause = throwable.getCause();
+ executeCallback.onFailure(null != cause ? cause : throwable);
}
}, executorService);
}
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngineTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngineTest.java
index dc6a816e387..0f7acc39f23 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngineTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/execute/ExecuteEngineTest.java
@@ -17,76 +17,111 @@
package org.apache.shardingsphere.data.pipeline.core.execute;
+import lombok.SneakyThrows;
import org.apache.shardingsphere.data.pipeline.api.executor.LifecycleExecutor;
-import org.apache.shardingsphere.data.pipeline.core.util.ThreadUtil;
import org.junit.Test;
-import java.util.Collections;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
-import static org.junit.Assert.assertFalse;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
public final class ExecuteEngineTest {
- @Test
- public void assertSubmitMoreThanMaxWorkerNumber() {
- ExecuteEngine executeEngine = ExecuteEngine.newFixedThreadInstance(2, ExecuteEngineTest.class.getSimpleName());
+ @Test(timeout = 30000)
+ public void assertSubmitAndTaskSucceeded() throws ExecutionException, InterruptedException {
+ LifecycleExecutor lifecycleExecutor = mock(LifecycleExecutor.class);
+ ExecuteCallback callback = mock(ExecuteCallback.class);
+ ExecuteEngine executeEngine = ExecuteEngine.newCachedThreadInstance(ExecuteEngineTest.class.getSimpleName());
+ Future<?> future = executeEngine.submit(lifecycleExecutor, callback);
+ future.get();
+ shutdownAndAwaitTerminal(executeEngine);
+ verify(lifecycleExecutor).run();
+ verify(callback).onSuccess();
+ }
+
+ @Test(timeout = 30000)
+ public void assertSubmitAndTaskFailed() {
+ LifecycleExecutor lifecycleExecutor = mock(LifecycleExecutor.class);
+ RuntimeException expectedException = new RuntimeException("Expected");
+ doThrow(expectedException).when(lifecycleExecutor).run();
+ ExecuteCallback callback = mock(ExecuteCallback.class);
+ ExecuteEngine executeEngine = ExecuteEngine.newCachedThreadInstance(ExecuteEngineTest.class.getSimpleName());
+ Future<?> future = executeEngine.submit(lifecycleExecutor, callback);
+ Throwable actualCause = null;
try {
- for (int i = 0; i < 5; i++) {
- Future<?> submit = executeEngine.submit(mockScalingExecutor());
- assertFalse(submit.isCancelled());
- }
- } catch (final RejectedExecutionException ex) {
+ future.get();
+ } catch (InterruptedException e) {
fail();
+ } catch (ExecutionException e) {
+ actualCause = e.getCause();
+ }
+ assertThat(actualCause, is(expectedException));
+ shutdownAndAwaitTerminal(executeEngine);
+ verify(callback).onFailure(expectedException);
+ }
+
+ @Test(timeout = 30000)
+ public void assertSubmitAllAndTasksSucceeded() throws ExecutionException, InterruptedException {
+ int taskCount = 8;
+ LifecycleExecutor lifecycleExecutor = mock(LifecycleExecutor.class);
+ List<LifecycleExecutor> lifecycleExecutors = new ArrayList<>(taskCount);
+ for (int i = 0; i < taskCount; i++) {
+ lifecycleExecutors.add(lifecycleExecutor);
}
+ ExecuteCallback callback = mock(ExecuteCallback.class);
+ ExecuteEngine executeEngine = ExecuteEngine.newFixedThreadInstance(2, ExecuteEngineTest.class.getSimpleName());
+ Future<?> future = executeEngine.submitAll(lifecycleExecutors, callback);
+ future.get();
+ shutdownAndAwaitTerminal(executeEngine);
+ verify(lifecycleExecutor, times(taskCount)).run();
+ verify(callback).onSuccess();
}
- @Test
- public void assertSubmitAllMoreThanMaxWorkerNumber() {
+ @Test(timeout = 30000)
+ public void assertSubmitAllAndOneOfTasksFailed() {
+ LifecycleExecutor lifecycleExecutor = mock(LifecycleExecutor.class);
+ List<LifecycleExecutor> lifecycleExecutors = new ArrayList<>(8);
+ for (int i = 0; i < 7; i++) {
+ lifecycleExecutors.add(lifecycleExecutor);
+ }
+ LifecycleExecutor failedExecutor = mock(LifecycleExecutor.class);
+ RuntimeException expectedException = new RuntimeException("Expected");
+ doThrow(expectedException).when(failedExecutor).run();
+ lifecycleExecutors.add(failedExecutor);
+ ExecuteCallback callback = mock(ExecuteCallback.class);
ExecuteEngine executeEngine = ExecuteEngine.newFixedThreadInstance(2, ExecuteEngineTest.class.getSimpleName());
+ Future<?> future = executeEngine.submitAll(lifecycleExecutors, callback);
+ Throwable actualCause = null;
try {
- for (int i = 0; i < 5; i++) {
- Future<?> submit = executeEngine.submitAll(Collections.singletonList(mockScalingExecutor()), mockExecuteCallback());
- assertFalse(submit.isCancelled());
- }
- } catch (final RejectedExecutionException ex) {
+ future.get();
+ } catch (InterruptedException e) {
fail();
+ } catch (ExecutionException e) {
+ actualCause = e.getCause();
}
+ assertThat(actualCause, is(expectedException));
+ shutdownAndAwaitTerminal(executeEngine);
+ verify(callback).onFailure(expectedException);
}
- private ExecuteCallback mockExecuteCallback() {
- return new ExecuteCallback() {
-
- @Override
- public void onSuccess() {
-
- }
-
- @Override
- public void onFailure(final Throwable throwable) {
-
- }
- };
- }
-
- private LifecycleExecutor mockScalingExecutor() {
-
- return new LifecycleExecutor() {
-
- @Override
- public void run() {
- ThreadUtil.sleep(100L);
- }
-
- @Override
- public void start() {
- }
-
- @Override
- public void stop() {
- }
- };
+ @SneakyThrows
+ private void shutdownAndAwaitTerminal(final ExecuteEngine executeEngine) {
+ Field field = ExecuteEngine.class.getDeclaredField("executorService");
+ field.setAccessible(true);
+ ExecutorService executorService = (ExecutorService) field.get(executeEngine);
+ executorService.shutdown();
+ executorService.awaitTermination(30, TimeUnit.SECONDS);
}
}