Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1632#discussion_r16323500
--- Diff:
core/src/main/scala/org/apache/spark/network/ConnectionManager.scala ---
@@ -652,19 +654,21 @@ 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 => {
+ if (isAckTimeout) {
+ logWarning(s"Ack message ${message.id} maybe received
after timeout")
+ }
--- End diff --
Before, it was undoubtedly a bug if we received an ack for a message and
didn't have a corresponding SentMessageStatus.
Now that we have timeouts, we have no way to distinguish between a
late-arriving ack for a SentMessageStatus that we've already deleted and a
bogus ack sent due to buggy code. As I commented upthread, one option would be
to simply convert this into a warning. But another option is to keep it as an
error _unless_ we've timed out at least once, in which case we treat it as a
warning.
---
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]
