You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2023/06/30 09:43:31 UTC
[flink] branch master updated (167d5640b24 -> 9d7b1663296)
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 167d5640b24 [FLINK-32467][rpc] Move common RPC utils to rpc-core
new 24e479571fa [FLINK-32495][connectors/common] Fix the bug that the shared thread factory causes the source alignment unit test to fail
new 6453a48e70f [hotfix] Migrate StreamExecutionEnvironmentTest to Junit5 and AssertJ
new 9d7b1663296 [FLINK-32495][connectors/common] Limit the CoordinatorExecutorThreadFactory to be a one-off
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:
.../coordinator/SourceCoordinatorProvider.java | 11 ++++-
.../coordinator/SourceCoordinatorContextTest.java | 3 +-
.../coordinator/SourceCoordinatorProviderTest.java | 52 +++++++++++++---------
.../coordinator/SourceCoordinatorTestBase.java | 15 +++----
4 files changed, 50 insertions(+), 31 deletions(-)
[flink] 02/03: [hotfix] Migrate StreamExecutionEnvironmentTest to Junit5 and AssertJ
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6453a48e70f542ee2618da10cb44a480e56fe0f3
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Fri Jun 30 13:32:02 2023 +0800
[hotfix] Migrate StreamExecutionEnvironmentTest to Junit5 and AssertJ
---
.../coordinator/SourceCoordinatorProviderTest.java | 36 ++++++++++------------
1 file changed, 16 insertions(+), 20 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
index 5452c4c0e06..348e3e048c7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
@@ -29,26 +29,24 @@ import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
import org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator;
import org.apache.flink.runtime.source.event.ReaderRegistrationEvent;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Unit tests for {@link SourceCoordinatorProvider}. */
-public class SourceCoordinatorProviderTest {
+class SourceCoordinatorProviderTest {
private static final OperatorID OPERATOR_ID = new OperatorID(1234L, 5678L);
private static final int NUM_SPLITS = 10;
private SourceCoordinatorProvider<MockSourceSplit> provider;
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
provider =
new SourceCoordinatorProvider<>(
"SourceCoordinatorProviderTest",
@@ -60,14 +58,14 @@ public class SourceCoordinatorProviderTest {
}
@Test
- public void testCreate() throws Exception {
+ void testCreate() throws Exception {
OperatorCoordinator coordinator =
provider.create(new MockOperatorCoordinatorContext(OPERATOR_ID, NUM_SPLITS));
- assertTrue(coordinator instanceof RecreateOnResetOperatorCoordinator);
+ assertThat(coordinator).isInstanceOf(RecreateOnResetOperatorCoordinator.class);
}
@Test
- public void testCheckpointAndReset() throws Exception {
+ void testCheckpointAndReset() throws Exception {
final OperatorCoordinator.Context context =
new MockOperatorCoordinatorContext(OPERATOR_ID, NUM_SPLITS);
final RecreateOnResetOperatorCoordinator coordinator =
@@ -94,19 +92,17 @@ public class SourceCoordinatorProviderTest {
coordinator.resetToCheckpoint(0L, bytes);
final SourceCoordinator<?, ?> restoredSourceCoordinator =
(SourceCoordinator<?, ?>) coordinator.getInternalCoordinator();
- assertNotEquals(
- "The restored source coordinator should be a different instance",
- restoredSourceCoordinator,
- sourceCoordinator);
+ assertThat(sourceCoordinator)
+ .as("The restored source coordinator should be a different instance")
+ .isNotEqualTo(restoredSourceCoordinator);
// FLINK-21452: do not (re)store registered readers
- assertEquals(
- "There should be no registered reader.",
- 0,
- restoredSourceCoordinator.getContext().registeredReaders().size());
+ assertThat(restoredSourceCoordinator.getContext().registeredReaders())
+ .as("There should be no registered reader.")
+ .isEmpty();
}
@Test
- public void testCallAsyncExceptionFailsJob() throws Exception {
+ void testCallAsyncExceptionFailsJob() throws Exception {
MockOperatorCoordinatorContext context =
new MockOperatorCoordinatorContext(OPERATOR_ID, NUM_SPLITS);
RecreateOnResetOperatorCoordinator coordinator =
[flink] 01/03: [FLINK-32495][connectors/common] Fix the bug that the shared thread factory causes the source alignment unit test to fail
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 24e479571facd07ff129c57cda86c7e34d4637bd
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Fri Jun 30 13:12:49 2023 +0800
[FLINK-32495][connectors/common] Fix the bug that the shared thread factory causes the source alignment unit test to fail
---
.../source/coordinator/SourceCoordinatorContextTest.java | 3 ++-
.../source/coordinator/SourceCoordinatorTestBase.java | 15 ++++++---------
2 files changed, 8 insertions(+), 10 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
index dc760aca544..70e2ffc33a4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
@@ -165,7 +165,8 @@ class SourceCoordinatorContextTest extends SourceCoordinatorTestBase {
new SourceCoordinatorContext<>(
coordinatorExecutorWithExceptionHandler,
manualWorkerExecutor,
- coordinatorThreadFactory,
+ new SourceCoordinatorProvider.CoordinatorExecutorThreadFactory(
+ coordinatorThreadName, operatorCoordinatorContext),
operatorCoordinatorContext,
new MockSourceSplitSerializer(),
splitSplitAssignmentTracker,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorTestBase.java
index fb2886f8fb0..8fba35df0d3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorTestBase.java
@@ -65,7 +65,7 @@ abstract class SourceCoordinatorTestBase {
protected MockOperatorCoordinatorContext operatorCoordinatorContext;
// ---- Mocks for the Source Coordinator Context ----
- protected SourceCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory;
+ protected String coordinatorThreadName;
protected SplitAssignmentTracker<MockSourceSplit> splitSplitAssignmentTracker;
protected SourceCoordinatorContext<MockSourceSplit> context;
@@ -82,10 +82,7 @@ abstract class SourceCoordinatorTestBase {
operatorCoordinatorContext =
new MockOperatorCoordinatorContext(TEST_OPERATOR_ID, NUM_SUBTASKS);
splitSplitAssignmentTracker = new SplitAssignmentTracker<>();
- String coordinatorThreadName = TEST_OPERATOR_ID.toHexString();
- coordinatorThreadFactory =
- new SourceCoordinatorProvider.CoordinatorExecutorThreadFactory(
- coordinatorThreadName, operatorCoordinatorContext);
+ coordinatorThreadName = TEST_OPERATOR_ID.toHexString();
sourceCoordinator = getNewSourceCoordinator();
context = sourceCoordinator.getContext();
@@ -215,14 +212,14 @@ abstract class SourceCoordinatorTestBase {
protected SourceCoordinatorContext<MockSourceSplit> getNewSourceCoordinatorContext()
throws Exception {
+ SourceCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory =
+ new SourceCoordinatorProvider.CoordinatorExecutorThreadFactory(
+ coordinatorThreadName, operatorCoordinatorContext);
SourceCoordinatorContext<MockSourceSplit> coordinatorContext =
new SourceCoordinatorContext<>(
Executors.newScheduledThreadPool(1, coordinatorThreadFactory),
Executors.newScheduledThreadPool(
- 1,
- new ExecutorThreadFactory(
- coordinatorThreadFactory.getCoordinatorThreadName()
- + "-worker")),
+ 1, new ExecutorThreadFactory(coordinatorThreadName + "-worker")),
coordinatorThreadFactory,
operatorCoordinatorContext,
new MockSourceSplitSerializer(),
[flink] 03/03: [FLINK-32495][connectors/common] Limit the CoordinatorExecutorThreadFactory to be a one-off
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9d7b1663296ef6f85572314a2b26a716981d8bff
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Fri Jun 30 13:18:00 2023 +0800
[FLINK-32495][connectors/common] Limit the CoordinatorExecutorThreadFactory to be a one-off
---
.../source/coordinator/SourceCoordinatorProvider.java | 11 ++++++++++-
.../coordinator/SourceCoordinatorProviderTest.java | 16 ++++++++++++++++
2 files changed, 26 insertions(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
index c76a3ec1910..d238e28be5f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
@@ -35,6 +35,8 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ThreadFactory;
import java.util.function.BiConsumer;
+import static org.apache.flink.util.Preconditions.checkState;
+
/** The provider of {@link SourceCoordinator}. */
public class SourceCoordinatorProvider<SplitT extends SourceSplit>
extends RecreateOnResetOperatorCoordinator.Provider {
@@ -94,7 +96,10 @@ public class SourceCoordinatorProvider<SplitT extends SourceSplit>
coordinatorListeningID);
}
- /** A thread factory class that provides some helper methods. */
+ /**
+ * A thread factory class that provides some helper methods. Because it is used to check the
+ * current thread, it is a one-off, do not use this ThreadFactory to create multiple threads.
+ */
public static class CoordinatorExecutorThreadFactory
implements ThreadFactory, Thread.UncaughtExceptionHandler {
@@ -130,6 +135,10 @@ public class SourceCoordinatorProvider<SplitT extends SourceSplit>
@Override
public synchronized Thread newThread(Runnable r) {
+ checkState(
+ t == null,
+ "Please using the new CoordinatorExecutorThreadFactory,"
+ + " this factory cannot new multiple threads.");
t = new Thread(r, coordinatorThreadName);
t.setContextClassLoader(cl);
t.setUncaughtExceptionHandler(this);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
index 348e3e048c7..11fefa5b981 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProviderTest.java
@@ -36,6 +36,7 @@ import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link SourceCoordinatorProvider}. */
class SourceCoordinatorProviderTest {
@@ -121,4 +122,19 @@ class SourceCoordinatorProviderTest {
Duration.ofSeconds(10L),
"The job did not fail before timeout.");
}
+
+ @Test
+ void testCoordinatorExecutorThreadFactoryNewMultipleThread() {
+ SourceCoordinatorProvider.CoordinatorExecutorThreadFactory
+ coordinatorExecutorThreadFactory =
+ new SourceCoordinatorProvider.CoordinatorExecutorThreadFactory(
+ "test_coordinator_thread",
+ new MockOperatorCoordinatorContext(
+ new OperatorID(1234L, 5678L), 3));
+
+ coordinatorExecutorThreadFactory.newThread(() -> {});
+ // coordinatorExecutorThreadFactory cannot create multiple threads.
+ assertThatThrownBy(() -> coordinatorExecutorThreadFactory.newThread(() -> {}))
+ .isInstanceOf(IllegalStateException.class);
+ }
}