[FLINK-10251] Handle oversized response messages in AkkaRpcActor #6876

yanghua · 2018-10-18T12:12:15Z

What is the purpose of the change

This pull request handles oversized response messages in AkkaRpcActor*

Brief change log

Handle oversized response messages in AkkaRpcActor

Verifying this change

TBD

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

yanghua · 2018-10-18T12:25:43Z

cc @tillrohrmann At the moment it seems that I don't see the infrastructure for testing AkkaRpcActor#handleRpcInvocation?

tillrohrmann

Thanks for your contribution @yanghua. I had some comments to further improve the PR.

I think we should add a test case verifying the behavior. We could for example add them to AkkaRpcActor and create an RpcEndpoint which creates a large response which exceeds the configured maximum frame size.

tillrohrmann · 2018-10-18T15:10:49Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java

@@ -108,9 +111,11 @@

 	private volatile boolean stopped;

-	public AkkaRpcService(final ActorSystem actorSystem, final Time timeout) {
+	public AkkaRpcService(final ActorSystem actorSystem, final Configuration configuration) {


Why did you pass the Configuration here? I think it would be better to revert this change.

Sorry, I forgot to refactor this code segment :

if (actorSystem.settings().config().hasPath(MAXIMUM_FRAME_SIZE_PATH)) { maximumFramesize = actorSystem.settings().config().getBytes(MAXIMUM_FRAME_SIZE_PATH); } else { // only local communication maximumFramesize = Long.MAX_VALUE; }

to use

configuration.getString(AkkaOptions.FRAMESIZE);

(this suggestion from @zentol)

If I would do this, I think it's better to replace scattered config items into one Configuration object so that we can encapsulate change?

Yes let's introduce an AkkaRpcServiceConfiguration.

tillrohrmann · 2018-10-18T15:14:17Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

+													"size, result size is " + resultSize + ", maximum frame size is " +
+													maximumFramesize + " ."));
+											} else {
+												promise.success(value);


I think it would be better to send the SerializedValue. That way we save some extra serialization work. This, of course, requires that the receiver expects a SerializedValue.

tillrohrmann · 2018-10-18T15:14:59Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

+										try {
+											resultSize = new SerializedValue(value).getByteArray().length;
+											if (resultSize > maximumFramesize) {
+												promise.failure(new Throwable("The result size exceeds the maximum " +


Let's throw an AkkaRpcException

tillrohrmann · 2018-10-18T15:15:37Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

+											resultSize = new SerializedValue(value).getByteArray().length;
+											if (resultSize > maximumFramesize) {
+												promise.failure(new Throwable("The result size exceeds the maximum " +
+													"size, result size is " + resultSize + ", maximum frame size is " +


Maybe it's enough to say The result size %s exceeds the maximum frame size %s.

tillrohrmann · 2018-10-18T15:16:23Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

+												promise.success(value);
+											}
+										} catch (IOException e) {
+											promise.failure(e);


Maybe wrap in AkkaRpcException("Failed to serialize the result for RPC call %s.", e)

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

yanghua · 2018-10-21T05:29:51Z

hi @tillrohrmann There is a problem about sending serialized data. For example, testNoSlotSharingAndBlockingResultReceiver will trigger allocateSlot rpc call, the result type is SingleLogicalSlot which has a field private final CompletableFuture<Void> releaseFuture. The field caused serialization error :

Caused by: org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException: Failed to serialize the result for RPC call : allocateSlot.
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.serializeResultAndVerifySize(AkkaRpcActor.java:335)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRpcInvocation$1(AkkaRpcActor.java:280)
	... 33 more
Caused by: java.io.NotSerializableException: java.util.concurrent.CompletableFuture

tillrohrmann · 2018-10-21T19:52:35Z

The problem is that you always try to serialize the result. However, it should only be serialized if the sender is remote.

yanghua · 2018-10-22T07:04:24Z

Thank you @tillrohrmann , I think I misunderstood your previous meaning about serialization results (I thought all the results were serialized). I have fixed it locally, but the Github service has failed (at least in China), and once the Github service is normal, I will push my changes.

yanghua · 2018-10-23T06:56:36Z

@tillrohrmann I have updated this PR, please review.

yanghua · 2018-10-23T11:20:50Z

@zentol Any suggestion?

yanghua · 2018-10-24T11:53:11Z

@tillrohrmann If you can give a review suggestion as soon as possible, then I can refactor it quickly so that it catches up with 1.7.0.

yanghua · 2018-10-29T13:07:56Z

@tillrohrmann Will you review this PR before the release of Flink 1.7.0?

tillrohrmann · 2018-11-02T10:23:41Z

Hi @yanghua sorry for my unresponsiveness. I think this fix won't make it into the 1.7 release anymore. But we will include in the next bug fix release!

yanghua · 2018-11-08T15:26:21Z

cc @tillrohrmann

tillrohrmann · 2018-11-09T13:05:49Z

Thanks for the reminder @yanghua. I'm still busy with the 1.7 release. Once it is out, I'll take a look at this PR.

yanghua · 2018-11-30T12:23:02Z

@tillrohrmann

tillrohrmann

Thanks for the update @yanghua. The changes go in a really good direction. I had some comments which we should address before merging.

tillrohrmann · 2019-01-09T10:38:38Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

+		final T rpcEndpoint,
+		final CompletableFuture<Boolean> terminationFuture,
+		final int version,
+		final long maximumFramesize) {


please indent parameters one level deeper.

tillrohrmann · 2019-01-09T10:38:54Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

 		this.rpcEndpoint = checkNotNull(rpcEndpoint, "rpc endpoint");
 		this.mainThreadValidator = new MainThreadValidatorUtil(rpcEndpoint);
 		this.terminationFuture = checkNotNull(terminationFuture);
 		this.version = version;
+		this.maximumFramesize = maximumFramesize;


should check that this is positive

tillrohrmann · 2019-01-09T10:41:44Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

@@ -71,7 +73,7 @@
 *
 * @param <T> Type of the {@link RpcEndpoint}
 */
-class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
+public class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {


Why does the AkkaRpcActor need to be public?

Oh, I created org.apache.flink.runtime.rpc.TestingRemoteAkkaRpcActor before. Now, I will move this class into pkg org.apache.flink.runtime.rpc.akka, then I can remove the public keyword.

tillrohrmann · 2019-01-09T10:42:23Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java

+							return  ((SerializedValue) o).deserializeValue(getClass().getClassLoader());
+						} catch (IOException e) {
+							throw new CompletionException(e);
+						} catch (ClassNotFoundException e) {


The two catch clause could be collapsed catch(IOException | ClassNotFoundException e)

tillrohrmann · 2019-01-09T10:45:34Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

+							Either<SerializedValue, AkkaRpcException> serializedResult =
+								serializeRemoteResultAndVerifySize(result, method.getName());
+							if (serializedResult.isLeft()) {
+								getSender().tell(serializedResult.left(), getSelf());


Serialized result should be wrapped in new Status.Success(result)

tillrohrmann · 2019-01-09T10:59:24Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java

+
+	static class OversizedResponseRpcEndpoint extends TestRpcEndpoint implements OversizedResponseMsgRpcGateway {
+
+		private volatile Byte[] bytes = new Byte[1024 * 10];


let's initialize the bytes via a constructor parameter. That way it is controllable and we don't need the getters and setters.

tillrohrmann · 2019-01-09T11:00:05Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java

+		}
+	}
+
+	static class TestingRemoteRpcService extends TestingRpcService {


No need for this RpcService if we properly simulate remote communication by using a second RpcService.

tillrohrmann · 2019-01-09T11:00:30Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MessageSerializationTest.java

-		akkaRpcService1 = new AkkaRpcService(actorSystem1, timeout);
-		akkaRpcService2 = new AkkaRpcService(actorSystem2, timeout);
+		akkaRpcService1 = new AkkaRpcService(actorSystem1, AkkaRpcServiceConfiguration.fromConfiguration(configuration));
+		akkaRpcService2 = new AkkaRpcService(actorSystem2, AkkaRpcServiceConfiguration.fromConfiguration(configuration));


we could reuse the first AkkaRpcServiceConfiguration.

tillrohrmann · 2019-01-09T11:02:34Z

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java

@@ -757,7 +758,7 @@ protected RpcService createRpcService(

 		final ActorSystem actorSystem = AkkaUtils.createActorSystem(effectiveAkkaConfig);

-		return new AkkaRpcService(actorSystem, askTimeout);
+		return new AkkaRpcService(actorSystem, AkkaRpcServiceConfiguration.fromConfiguration(configuration));


I would suggest to replace the askTimeout parameter with a AkkaRpcServiceConfiguration akkaRpcServiceConfiguration parameter and pass it to this method.

tillrohrmann · 2019-01-09T11:05:15Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java

+						return o;
+					}
+				});
+			}


This block should go into the invokeRpc method because we also need to unwrap the synchronous calls.

We should also add a test case that synchronous calls (Rpc's which return directly a value instead of a CompletableFuture) work with this change as well.

yanghua · 2019-01-10T08:20:38Z

@tillrohrmann I have updated this PR.

tillrohrmann

Thanks for the update @yanghua. The code looks now really good. I had some minor comments. After resolving them, we should be good to merge this PR.

tillrohrmann · 2019-01-16T11:01:31Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java

+				try {
+					result = ((SerializedValue) returnedResult).deserializeValue(getClass().getClassLoader());
+				} catch (IOException | ClassNotFoundException e) {
+					throw new CompletionException(e);


This should not be a CompletionException. Instead I would throw a RpcException saying that one could not deserialize a serialized payload.

tillrohrmann · 2019-01-16T11:02:01Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java

+					try {
+						return  ((SerializedValue) o).deserializeValue(getClass().getClassLoader());
+					} catch (IOException | ClassNotFoundException e) {
+						throw new CompletionException(e);


The same applies here. It would be good CompletionExcpetion(new RpcException("Could not deserialize...", e));

tillrohrmann · 2019-01-16T11:03:10Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java

+				}
+			} else {
+				result = returnedResult;
+			}


This is somewhat a code duplication with the other branch. Could we deduplicate this code by applying the deserialization logic on the future? Additionally it would be good to catch ExecutionException when calling resultFuture.get() and unstrip them via ExceptionUtils.unstripExecutionException.

I did not find ExceptionUtils#unstripExecutionException, but exists ExceptionUtils#stripExecutionException. Did you mean ExceptionUtils#rethrow?

Hi @tillrohrmann I have updated the PR based on your suggestion, except this one. I did not find the method you mentioned.

Oops, I meant stripExecutionException. I'll take a look and see whether I can do it while merging.

tillrohrmann · 2019-01-16T11:05:37Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceConfiguration.java

+/**
+ * Configuration object for {@link AkkaRpcService}.
+ */
+public class AkkaRpcServiceConfiguration {


We could also add the Configuration to this class. Then we only need to pass in this object instead of the configuration and the object.

tillrohrmann · 2019-01-16T11:06:11Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceConfiguration.java

+
+	public AkkaRpcServiceConfiguration(Time timeout, long maximumFramesize) {
+		this.timeout = timeout;
+		this.maximumFramesize = maximumFramesize;


Check that timeout is not null and maximumFramesize > 0 missing.

tillrohrmann · 2019-01-16T11:06:48Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceUtils.java

+		String akkaConfigStr = String.format(SIMPLE_AKKA_CONFIG_TEMPLATE, maxFrameSizeStr);
+		Config akkaConfig = ConfigFactory.parseString(akkaConfigStr);
+		return akkaConfig.getBytes(MAXIMUM_FRAME_SIZE_PATH);
+	}


tillrohrmann · 2019-01-16T11:11:01Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

 	private State state;

-	AkkaRpcActor(final T rpcEndpoint, final CompletableFuture<Boolean> terminationFuture, final int version) {
+	public AkkaRpcActor(


This one is still public. Can we make it package private again?

tillrohrmann · 2019-01-16T11:11:57Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/SyncCallsTest.java

+				rpcEndpoint.shutDown();
+			}
+		}
+


line break not needed.

tillrohrmann · 2019-01-16T11:12:24Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/SyncCallsTest.java

+/**
+ * RPC sync invoke test.
+ */
+public class SyncCallsTest extends TestLogger {


Nice tests. Can we also add a synchronous call with an oversized message?

tillrohrmann

Thanks for the updates @yanghua. LGTM. Merging this PR.

This commit slightly refactors and cleans up the handling of over sized messages. This closes apache#6876.

yanghua force-pushed the FLINK-10251 branch 2 times, most recently from e92b8af to e3c1155 Compare October 18, 2018 12:20

tillrohrmann requested changes Oct 18, 2018

View reviewed changes

yanghua force-pushed the FLINK-10251 branch 3 times, most recently from 77e882d to 8d3e3fb Compare October 20, 2018 09:56

yanghua force-pushed the FLINK-10251 branch from 8d3e3fb to 717b244 Compare October 22, 2018 03:27

yanghua force-pushed the FLINK-10251 branch from 717b244 to 8c2496e Compare October 22, 2018 16:59

tillrohrmann requested changes Jan 9, 2019

View reviewed changes

yanghua force-pushed the FLINK-10251 branch from 8c2496e to 22063e1 Compare January 10, 2019 03:16

tillrohrmann requested changes Jan 16, 2019

View reviewed changes

tillrohrmann self-assigned this Jan 17, 2019

tillrohrmann approved these changes Jan 23, 2019

View reviewed changes

tillrohrmann added a commit to tillrohrmann/flink that referenced this pull request Jan 24, 2019

[FLINK-10251][rpc] Refactor handling of over sized messages

287bdee

This commit slightly refactors and cleans up the handling of over sized messages. This closes apache#6876.

tillrohrmann added a commit to tillrohrmann/flink that referenced this pull request Jan 24, 2019

[FLINK-10251][rpc] Refactor handling of over sized messages

946ba91

This commit slightly refactors and cleans up the handling of over sized messages. This closes apache#6876.

yanghua added 2 commits January 24, 2019 16:08

[FLINK-10251] Handle oversized response messages in AkkaRpcActor

ecb8481

do not serialize local response

8af46e3

yanghua added 2 commits January 24, 2019 16:08

refactor code and added some test case for rpc sync call

6d33fdb

refactor code based on review suggestion

227fa97

yanghua force-pushed the FLINK-10251 branch from 41557e8 to 227fa97 Compare January 24, 2019 08:09

asfgit closed this in 3667ea6 Jan 24, 2019

rmetzger added the component=Runtime/Coordination label Mar 18, 2019


		static class OversizedResponseRpcEndpoint extends TestRpcEndpoint implements OversizedResponseMsgRpcGateway {

		private volatile Byte[] bytes = new Byte[1024 * 10];

[FLINK-10251] Handle oversized response messages in AkkaRpcActor #6876

[FLINK-10251] Handle oversized response messages in AkkaRpcActor #6876

Conversation

yanghua commented Oct 18, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

yanghua commented Oct 18, 2018

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yanghua commented Oct 21, 2018

tillrohrmann commented Oct 21, 2018

yanghua commented Oct 22, 2018

yanghua commented Oct 23, 2018

yanghua commented Oct 23, 2018

yanghua commented Oct 24, 2018

yanghua commented Oct 29, 2018

tillrohrmann commented Nov 2, 2018

yanghua commented Nov 8, 2018

tillrohrmann commented Nov 9, 2018

yanghua commented Nov 30, 2018

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yanghua commented Jan 10, 2019

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment