Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1632#discussion_r16095159
--- Diff:
core/src/main/scala/org/apache/spark/network/ConnectionManager.scala ---
@@ -652,19 +655,25 @@ private[spark] class ConnectionManager(
}
}
if (bufferMessage.hasAckId()) {
- val sentMessageStatus = messageStatuses.synchronized {
+ messageStatuses.synchronized {
messageStatuses.get(bufferMessage.ackId) match {
case Some(status) => {
messageStatuses -= bufferMessage.ackId
- status
+ status.markDone(Some(message))
}
case None => {
- throw new Exception("Could not find reference for received
ack message " +
- message.id)
+ /**
+ * If isAckTimeout == true, Future returned from
sendMessageReliably should fail
+ * and finally, FetchFailedException is thrown so in this
case, we don't need
+ * to throw Exception here
+ */
+ if (!isAckTimeout) {
--- End diff --
Since we delete the MessageStatuses of messages that fail with timeouts, I
guess we have no way to distinguish between an ACK for a message that we didn't
send and an ACK for a timed-out message. I guess that `isAckTimeout` is used
to strike a reasonable compromise in which we'll detect errors if no timeouts
occur.
It might be a good idea to `logWarning` if we receive an ACK after we've
timed out; this might help when debugging: if the ack timeout is set too low,
warning messages would appear in the logs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
