Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/3647#issuecomment-67402902
Hmm, so it looks like `reportError` may create a new entry in the
`receiverInfo` map if it processes an error from a previously-unknown receiver:
```scala
/** Report error sent by a receiver */
private def reportError(streamId: Int, message: String, error: String) {
val newReceiverInfo = receiverInfo.get(streamId) match {
case Some(oldInfo) =>
oldInfo.copy(lastErrorMessage = message, lastError = error)
case None =>
logWarning("No prior receiver info")
ReceiverInfo(streamId, "", null, false, "", lastErrorMessage =
message, lastError = error)
}
receiverInfo(streamId) = newReceiverInfo
listenerBus.post(StreamingListenerReceiverError(receiverInfo(streamId)))
val messageWithError = if (error != null && !error.isEmpty) {
s"$message - $error"
} else {
s"$message"
}
logWarning(s"Error reported by receiver for stream $streamId:
$messageWithError")
}
```
This means that we'll leak `receiverInfo` map entires if we ever receive
errors after removing receivers. I don't think that this should ever happen,
though. If we're concerned about this, I suppose we could rewrite this method
to generate the new receiver info, post the new info to the listener bus, then
store the info in the mapping only if there was an old entry.
---
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]
