You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/02/13 08:27:10 UTC
[flink] branch release-1.17 updated (19c05ef0c86 -> a5536f2f2f5)
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a change to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
from 19c05ef0c86 [FLINK-30969][python] Remove the __init__.py files under the examples
new fa85ca20ac5 [hotfix] Migrate ExecutionGraphPartitionReleaseTest to Junit5 and AssertJ.
new 8e8aa2aac16 [hotfix] Introduce Junit5 extension for TestingComponentMainThreadExecutor.
new a5536f2f2f5 [FLINK-30944] Fix thread leak in ExecutionGraphPartitionReleaseTest.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../ExecutionGraphPartitionReleaseTest.java | 70 ++++++++++------------
.../TestingComponentMainThreadExecutor.java | 39 ++++++++++++
2 files changed, 70 insertions(+), 39 deletions(-)
[flink] 01/03: [hotfix] Migrate ExecutionGraphPartitionReleaseTest to Junit5 and AssertJ.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fa85ca20ac5919c52a548a069551d04b00885db1
Author: Weijie Guo <re...@163.com>
AuthorDate: Wed Feb 8 11:26:26 2023 +0800
[hotfix] Migrate ExecutionGraphPartitionReleaseTest to Junit5 and AssertJ.
---
.../ExecutionGraphPartitionReleaseTest.java | 56 ++++++++++------------
1 file changed, 25 insertions(+), 31 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
index 29711bc9956..5ff3af15b9e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
@@ -33,31 +33,27 @@ import org.apache.flink.runtime.scheduler.SchedulerBase;
import org.apache.flink.runtime.scheduler.SchedulerTestingUtils;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.core.IsEqual.equalTo;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for the interactions of the {@link ExecutionGraph} and {@link
* PartitionGroupReleaseStrategy}.
*/
-public class ExecutionGraphPartitionReleaseTest extends TestLogger {
+class ExecutionGraphPartitionReleaseTest {
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ public static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_EXTENSION =
+ TestingUtils.defaultExecutorExtension();
private static final ScheduledExecutorService scheduledExecutorService =
Executors.newSingleThreadScheduledExecutor();
@@ -67,7 +63,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
scheduledExecutorService));
@Test
- public void testStrategyNotifiedOfFinishedVerticesAndResultsRespected() throws Exception {
+ void testStrategyNotifiedOfFinishedVerticesAndResultsRespected() throws Exception {
// setup a simple pipeline of 3 operators with blocking partitions
final JobVertex sourceVertex = ExecutionGraphTestUtils.createNoOpVertex(1);
final JobVertex operatorVertex = ExecutionGraphTestUtils.createNoOpVertex(1);
@@ -98,7 +94,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
scheduler.updateTaskExecutionState(
new TaskExecutionState(
sourceExecution.getAttemptId(), ExecutionState.FINISHED));
- assertThat(releasedPartitions, empty());
+ assertThat(releasedPartitions).isEmpty();
});
mainThreadExecutor.execute(
@@ -110,10 +106,9 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
scheduler.updateTaskExecutionState(
new TaskExecutionState(
operatorExecution.getAttemptId(), ExecutionState.FINISHED));
- assertThat(releasedPartitions, hasSize(1));
- assertThat(
- releasedPartitions.remove(),
- equalTo(
+ assertThat(releasedPartitions).hasSize(1);
+ assertThat(releasedPartitions.remove())
+ .isEqualTo(
new ResultPartitionID(
sourceExecution
.getVertex()
@@ -121,7 +116,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
.keySet()
.iterator()
.next(),
- sourceExecution.getAttemptId())));
+ sourceExecution.getAttemptId()));
});
mainThreadExecutor.execute(
@@ -133,10 +128,9 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
new TaskExecutionState(
sinkExecution.getAttemptId(), ExecutionState.FINISHED));
- assertThat(releasedPartitions, hasSize(1));
- assertThat(
- releasedPartitions.remove(),
- equalTo(
+ assertThat(releasedPartitions).hasSize(1);
+ assertThat(releasedPartitions.remove())
+ .isEqualTo(
new ResultPartitionID(
operatorExecution
.getVertex()
@@ -144,12 +138,12 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
.keySet()
.iterator()
.next(),
- operatorExecution.getAttemptId())));
+ operatorExecution.getAttemptId()));
});
}
@Test
- public void testStrategyNotifiedOfUnFinishedVertices() throws Exception {
+ void testStrategyNotifiedOfUnFinishedVertices() throws Exception {
// setup a pipeline of 2 failover regions (f1 -> f2), where
// f1 is just a source
// f2 consists of 3 operators (o1,o2,o3), where o1 consumes f1, and o2/o3 consume o1
@@ -190,7 +184,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
scheduler.updateTaskExecutionState(
new TaskExecutionState(
sourceExecution.getAttemptId(), ExecutionState.FINISHED));
- assertThat(releasedPartitions, empty());
+ assertThat(releasedPartitions).isEmpty();
});
mainThreadExecutor.execute(
@@ -202,7 +196,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
scheduler.updateTaskExecutionState(
new TaskExecutionState(
operator1Execution.getAttemptId(), ExecutionState.FINISHED));
- assertThat(releasedPartitions, empty());
+ assertThat(releasedPartitions).isEmpty();
});
mainThreadExecutor.execute(
@@ -214,7 +208,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
scheduler.updateTaskExecutionState(
new TaskExecutionState(
operator2Execution.getAttemptId(), ExecutionState.FINISHED));
- assertThat(releasedPartitions, empty());
+ assertThat(releasedPartitions).isEmpty();
});
mainThreadExecutor.execute(
@@ -223,7 +217,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
getCurrentExecution(operator2Vertex, executionGraph);
// reset o2
operator2Execution.getVertex().resetForNewExecution();
- assertThat(releasedPartitions, empty());
+ assertThat(releasedPartitions).isEmpty();
});
mainThreadExecutor.execute(
@@ -234,7 +228,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
scheduler.updateTaskExecutionState(
new TaskExecutionState(
operator3Execution.getAttemptId(), ExecutionState.FINISHED));
- assertThat(releasedPartitions, empty());
+ assertThat(releasedPartitions).isEmpty();
});
}
@@ -255,7 +249,7 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
new DefaultSchedulerBuilder(
jobGraph,
mainThreadExecutor.getMainThreadExecutor(),
- EXECUTOR_RESOURCE.getExecutor())
+ EXECUTOR_EXTENSION.getExecutor())
.setExecutionSlotAllocatorFactory(
SchedulerTestingUtils.newSlotSharingExecutionSlotAllocatorFactory())
.setPartitionTracker(partitionTracker)
[flink] 02/03: [hotfix] Introduce Junit5 extension for TestingComponentMainThreadExecutor.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8e8aa2aac165620529ef674e7dc5d8e39ddfeae0
Author: Weijie Guo <re...@163.com>
AuthorDate: Wed Feb 8 11:42:11 2023 +0800
[hotfix] Introduce Junit5 extension for TestingComponentMainThreadExecutor.
---
.../TestingComponentMainThreadExecutor.java | 39 ++++++++++++++++++++++
1 file changed, 39 insertions(+)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java
index 10d4b17ec99..c767fc14d1b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java
@@ -25,6 +25,9 @@ import org.apache.flink.util.function.FunctionUtils;
import org.apache.flink.util.function.SupplierWithException;
import org.apache.flink.util.function.ThrowingRunnable;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
import org.junit.rules.ExternalResource;
import javax.annotation.Nonnull;
@@ -103,4 +106,40 @@ public class TestingComponentMainThreadExecutor {
return componentMainThreadTestExecutor;
}
}
+
+ /** Test extension for convenience. */
+ public static class Extension implements BeforeAllCallback, AfterAllCallback {
+ private final long shutdownTimeoutMillis;
+
+ private TestingComponentMainThreadExecutor componentMainThreadTestExecutor;
+
+ private ScheduledExecutorService innerExecutorService;
+
+ public Extension() {
+ this(500L);
+ }
+
+ public Extension(long shutdownTimeoutMillis) {
+ this.shutdownTimeoutMillis = shutdownTimeoutMillis;
+ }
+
+ @Override
+ public void beforeAll(ExtensionContext extensionContext) throws Exception {
+ this.innerExecutorService = Executors.newSingleThreadScheduledExecutor();
+ this.componentMainThreadTestExecutor =
+ new TestingComponentMainThreadExecutor(
+ ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
+ innerExecutorService));
+ }
+
+ @Override
+ public void afterAll(ExtensionContext extensionContext) throws Exception {
+ ExecutorUtils.gracefulShutdown(
+ shutdownTimeoutMillis, TimeUnit.MILLISECONDS, innerExecutorService);
+ }
+
+ public TestingComponentMainThreadExecutor getComponentMainThreadTestExecutor() {
+ return componentMainThreadTestExecutor;
+ }
+ }
}
[flink] 03/03: [FLINK-30944] Fix thread leak in ExecutionGraphPartitionReleaseTest.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a5536f2f2f5b0574d8e8ecb768afd049667a1fba
Author: Weijie Guo <re...@163.com>
AuthorDate: Wed Feb 8 11:43:35 2023 +0800
[FLINK-30944] Fix thread leak in ExecutionGraphPartitionReleaseTest.
---
.../executiongraph/ExecutionGraphPartitionReleaseTest.java | 14 ++++++--------
1 file changed, 6 insertions(+), 8 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
index 5ff3af15b9e..ca9fbfbb1ab 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
@@ -17,7 +17,6 @@
package org.apache.flink.runtime.executiongraph;
-import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionGroupReleaseStrategy;
import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
@@ -40,7 +39,6 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.ArrayDeque;
import java.util.Queue;
-import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import static org.assertj.core.api.Assertions.assertThat;
@@ -55,12 +53,12 @@ class ExecutionGraphPartitionReleaseTest {
public static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_EXTENSION =
TestingUtils.defaultExecutorExtension();
- private static final ScheduledExecutorService scheduledExecutorService =
- Executors.newSingleThreadScheduledExecutor();
- private static final TestingComponentMainThreadExecutor mainThreadExecutor =
- new TestingComponentMainThreadExecutor(
- ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
- scheduledExecutorService));
+ @RegisterExtension
+ public static final TestingComponentMainThreadExecutor.Extension MAIN_THREAD_EXTENSION =
+ new TestingComponentMainThreadExecutor.Extension();
+
+ private final TestingComponentMainThreadExecutor mainThreadExecutor =
+ MAIN_THREAD_EXTENSION.getComponentMainThreadTestExecutor();
@Test
void testStrategyNotifiedOfFinishedVerticesAndResultsRespected() throws Exception {