You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/07/27 18:49:56 UTC
[flink] branch master updated (7ac1fb9 -> abf9752)
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 7ac1fb9 [FLINK-23500][akka] Create tmp dir if it doesn't exist
new d387c30 [hotfix] Move DirectlyFailingFatalErrorHandler to flink-runtime
new abf9752 [FLINK-23504][tests] Isolate TestingRpcService from Akka
The 2 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:
flink-rpc/flink-rpc-akka-loader/pom.xml | 11 +++
.../flink/runtime/rpc/akka/AkkaRpcActorTest.java | 7 +-
flink-runtime/pom.xml | 15 ----
.../rpc/DirectlyFailingFatalErrorHandler.java | 0
.../flink/runtime/rpc/TestingRpcService.java | 89 ++++++++++++++++++----
pom.xml | 18 +++++
6 files changed, 108 insertions(+), 32 deletions(-)
rename {flink-rpc/flink-rpc-akka => flink-runtime}/src/test/java/org/apache/flink/runtime/rpc/DirectlyFailingFatalErrorHandler.java (100%)
rename {flink-rpc/flink-rpc-akka => flink-runtime}/src/test/java/org/apache/flink/runtime/rpc/TestingRpcService.java (69%)
[flink] 01/02: [hotfix] Move DirectlyFailingFatalErrorHandler to
flink-runtime
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d387c3052af56bb15dd66535c4ef294e6d78aab3
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Jul 26 20:33:03 2021 +0200
[hotfix] Move DirectlyFailingFatalErrorHandler to flink-runtime
---
.../org/apache/flink/runtime/rpc/DirectlyFailingFatalErrorHandler.java | 0
1 file changed, 0 insertions(+), 0 deletions(-)
diff --git a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/DirectlyFailingFatalErrorHandler.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/DirectlyFailingFatalErrorHandler.java
similarity index 100%
rename from flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/DirectlyFailingFatalErrorHandler.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/rpc/DirectlyFailingFatalErrorHandler.java
[flink] 02/02: [FLINK-23504][tests] Isolate TestingRpcService from
Akka
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit abf9752c9de3810e10b9f7bbb937cbdf288d2789
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Jul 26 20:40:22 2021 +0200
[FLINK-23504][tests] Isolate TestingRpcService from Akka
---
flink-rpc/flink-rpc-akka-loader/pom.xml | 11 +++
.../flink/runtime/rpc/akka/AkkaRpcActorTest.java | 7 +-
flink-runtime/pom.xml | 15 ----
.../flink/runtime/rpc/TestingRpcService.java | 89 ++++++++++++++++++----
pom.xml | 18 +++++
5 files changed, 108 insertions(+), 32 deletions(-)
diff --git a/flink-rpc/flink-rpc-akka-loader/pom.xml b/flink-rpc/flink-rpc-akka-loader/pom.xml
index 10bb645..07528d3 100644
--- a/flink-rpc/flink-rpc-akka-loader/pom.xml
+++ b/flink-rpc/flink-rpc-akka-loader/pom.xml
@@ -127,6 +127,17 @@ under the License.
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>forbid-direct-akka-rpc-dependencies</id>
+ <!-- This module needs the dependency in order to bundle flink-rpc-akka. -->
+ <phase>none</phase>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
diff --git a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java b/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
index b522d44..e99b571 100644
--- a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
+++ b/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.rpc.akka;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.FlinkMatchers;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils;
@@ -28,7 +29,6 @@ import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcServiceUtils;
import org.apache.flink.runtime.rpc.RpcTimeout;
import org.apache.flink.runtime.rpc.RpcUtils;
-import org.apache.flink.runtime.rpc.TestingRpcService;
import org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException;
import org.apache.flink.runtime.rpc.exceptions.RecipientUnreachableException;
import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
@@ -87,7 +87,10 @@ public class AkkaRpcActorTest extends TestLogger {
@BeforeClass
public static void setup() {
- akkaRpcService = new TestingRpcService();
+ akkaRpcService =
+ new AkkaRpcService(
+ AkkaUtils.createLocalActorSystem(new Configuration()),
+ AkkaRpcServiceConfiguration.defaultConfiguration());
}
@AfterClass
diff --git a/flink-runtime/pom.xml b/flink-runtime/pom.xml
index dc50cb1..b5217fb 100644
--- a/flink-runtime/pom.xml
+++ b/flink-runtime/pom.xml
@@ -174,21 +174,6 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-rpc-akka</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-rpc-akka</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- <type>test-jar</type>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-netty-tcnative-dynamic</artifactId>
<scope>test</scope>
</dependency>
diff --git a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/TestingRpcService.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingRpcService.java
similarity index 69%
rename from flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/TestingRpcService.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingRpcService.java
index b84e6fc..401b829 100644
--- a/flink-rpc/flink-rpc-akka/src/test/java/org/apache/flink/runtime/rpc/TestingRpcService.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingRpcService.java
@@ -19,14 +19,15 @@
package org.apache.flink.runtime.rpc;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
-import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceConfiguration;
-import org.apache.flink.runtime.rpc.akka.AkkaUtils;
import org.apache.flink.util.concurrent.FutureUtils;
+import org.apache.flink.util.concurrent.ScheduledExecutor;
import java.io.Serializable;
+import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -51,7 +52,11 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* verify(testGateway, timeout(1000)).theTestMethod(any(UUID.class), anyString());
* }</pre>
*/
-public class TestingRpcService extends AkkaRpcService {
+public class TestingRpcService implements RpcService {
+
+ // load RpcSystem once to save initialization costs
+ // this is safe because it is state-less
+ private static final RpcSystem RPC_SYSTEM_SINGLETON = RpcSystem.load();
private static final Function<RpcGateway, CompletableFuture<RpcGateway>>
DEFAULT_RPC_GATEWAY_FUTURE_FUNCTION = CompletableFuture::completedFuture;
@@ -62,16 +67,16 @@ public class TestingRpcService extends AkkaRpcService {
private volatile Function<RpcGateway, CompletableFuture<RpcGateway>> rpcGatewayFutureFunction =
DEFAULT_RPC_GATEWAY_FUTURE_FUNCTION;
- /** Creates a new {@code TestingRpcService}. */
- public TestingRpcService() {
- this(new Configuration());
- }
+ private final RpcService backingRpcService;
/** Creates a new {@code TestingRpcService}, using the given configuration. */
- public TestingRpcService(Configuration configuration) {
- super(
- AkkaUtils.createLocalActorSystem(configuration),
- AkkaRpcServiceConfiguration.fromConfiguration(configuration));
+ public TestingRpcService() {
+ try {
+ this.backingRpcService =
+ RPC_SYSTEM_SINGLETON.localServiceBuilder(new Configuration()).createAndStart();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
this.registeredConnections = new ConcurrentHashMap<>();
}
@@ -80,7 +85,7 @@ public class TestingRpcService extends AkkaRpcService {
@Override
public CompletableFuture<Void> stopService() {
- final CompletableFuture<Void> terminationFuture = super.stopService();
+ final CompletableFuture<Void> terminationFuture = backingRpcService.stopService();
terminationFuture.whenComplete(
(Void ignored, Throwable throwable) -> {
@@ -126,7 +131,7 @@ public class TestingRpcService extends AkkaRpcService {
+ clazz));
}
} else {
- return super.connect(address, clazz);
+ return backingRpcService.connect(address, clazz);
}
}
@@ -149,7 +154,7 @@ public class TestingRpcService extends AkkaRpcService {
+ clazz));
}
} else {
- return super.connect(address, fencingToken, clazz);
+ return backingRpcService.connect(address, fencingToken, clazz);
}
}
@@ -165,4 +170,58 @@ public class TestingRpcService extends AkkaRpcService {
Function<RpcGateway, CompletableFuture<RpcGateway>> rpcGatewayFutureFunction) {
this.rpcGatewayFutureFunction = rpcGatewayFutureFunction;
}
+
+ // ------------------------------------------------------------------------
+ // simple wrappers
+ // ------------------------------------------------------------------------
+
+ @Override
+ public String getAddress() {
+ return backingRpcService.getAddress();
+ }
+
+ @Override
+ public int getPort() {
+ return backingRpcService.getPort();
+ }
+
+ @Override
+ public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
+ return backingRpcService.startServer(rpcEndpoint);
+ }
+
+ @Override
+ public <F extends Serializable> RpcServer fenceRpcServer(RpcServer rpcServer, F fencingToken) {
+ return backingRpcService.fenceRpcServer(rpcServer, fencingToken);
+ }
+
+ @Override
+ public void stopServer(RpcServer selfGateway) {
+ backingRpcService.stopServer(selfGateway);
+ }
+
+ @Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return backingRpcService.getTerminationFuture();
+ }
+
+ @Override
+ public ScheduledExecutor getScheduledExecutor() {
+ return backingRpcService.getScheduledExecutor();
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleRunnable(Runnable runnable, long delay, TimeUnit unit) {
+ return backingRpcService.scheduleRunnable(runnable, delay, unit);
+ }
+
+ @Override
+ public void execute(Runnable runnable) {
+ backingRpcService.execute(runnable);
+ }
+
+ @Override
+ public <T> CompletableFuture<T> execute(Callable<T> callable) {
+ return backingRpcService.execute(callable);
+ }
}
diff --git a/pom.xml b/pom.xml
index 12ca18f..69eba74 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1655,6 +1655,24 @@ under the License.
</configuration>
</execution>
<execution>
+ <id>forbid-direct-akka-rpc-dependencies</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <rules>
+ <bannedDependencies>
+ <excludes>
+ <exclude>org.apache.flink:flink-rpc-akka</exclude>
+ </excludes>
+ <message>
+ Direct dependencies on flink-rpc-akka are not allowed. Depend on flink-rpc-akka-loader instead, and use RpcSystem#load or the TestingRpcService.
+ </message>
+ </bannedDependencies>
+ </rules>
+ </configuration>
+ </execution>
+ <execution>
<id>dependency-convergence</id>
<!-- disabled by default as it interacts badly with shade-plugin -->
<phase>none</phase>