[SPARK-11098][Core]Add Outbox to cache the sending messages to resolve the message disorder issue #9197
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
Test build #44060 has finished for PR 9197 at commit
|
|
Test build #44062 has finished for PR 9197 at commit
|
|
retest this please |
|
Test build #44072 has finished for PR 9197 at commit
|
|
|
||
| /** | ||
| * A map for [[RpcAddress]] and [[Outbox]]. When we are connecting to a remote [[RpcAddress]], | ||
| * we just put messages to its [[Outbox]] to implement a non-block `send` method. |
|
I was hoping we could somehow avoid needing an outbox, but I guess that's the easiest way to go. The code LGTM, although I'll probably take another look. |
|
Test build #44114 has finished for PR 9197 at commit
|
|
|
||
| override def onSuccess(response: Array[Byte]): Unit = { | ||
| val ack = deserialize[Ack](response) | ||
| logDebug(s"Receive ack from ${ack.sender}") |
There was a problem hiding this comment.
is this log useful? maybe we don't need to do much here ...
|
LGTM. @vanzin if you have time to take a closer look, that'd be great. |
| // update messages and it's safe to just drain the queue. | ||
| var message = messages.poll() | ||
| while (message != null) { | ||
| message.callback.onFailure(new SparkException("Message is dropped because Outbox is stopped")) |
There was a problem hiding this comment.
The SparkException can be constructed outside the while loop and reused.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
The current NettyRpc has a message order issue because it uses a thread pool to send messages. E.g., running the following two lines in the same thread,
The remote endpoint may see "B" before "A" because sending "A" and "B" are in parallel.
To resolve this issue, this PR added an outbox for each connection, and if we are connecting to the remote node when sending messages, just cache the sending messages in the outbox and send them one by one when the connection is established.