Github user mridulm commented on a diff in the pull request:
https://github.com/apache/spark/pull/2019#discussion_r16632025
--- Diff: core/src/main/scala/org/apache/spark/network/Connection.scala ---
@@ -118,14 +118,33 @@ abstract class Connection(val channel: SocketChannel,
val selector: Selector,
}
def close() {
- closed = true
- val k = key()
- if (k != null) {
- k.cancel()
+ synchronized {
+ /**
+ * We should avoid executing closing sequence
+ * double by a same thread.
+ * Otherwise we can fail to call connectionsById.get() in
+ * ConnectionManager#removeConnection() at the 2nd time
+ */
+ if (!closed) {
+ disposeSasl()
+
+ /**
+ * callOnCloseCallback() should be invoked
+ * before k.cancel() and channel.close()
+ * to avoid key() returns null.
+ * If key() returns null before callOnCloseCallback(),
+ * We cannot remove entry from connectionsByKey in
ConnectionManager
+ * and end up being threw CancelledKeyException.
+ */
+ callOnCloseCallback()
+ val k = key()
+ if (k != null) {
+ k.cancel()
+ }
+ channel.close()
+ closed = true
+ }
--- End diff --
I think you are misunderstanding the intent of what close is supposed to do
for Connection classes. It is supposed to mirror normal expectation of close on
streams - barring the bug I mentioned about.
In a nutshell, it is supposed to mark connection as closed (so the repeated
invocations of the method are idempotent), and cleanup if required. Take a look
at how close is implemented in general in various jdk IO classes.
---
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]
