[kafka] branch 2.0 updated: KAFKA-7464; catch exceptions in "leaderEndpoint.close()" when shutting down ReplicaFetcherThread

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lindong pushed a commit to branch 2.0
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.0 by this push:
     new 6845abc  KAFKA-7464; catch exceptions in "leaderEndpoint.close()" when shutting down ReplicaFetcherThread
6845abc is described below

commit 6845abc05fe12447a1dbd95d10abd600763a0ccc
Author: Zhanxiang (Patrick) Huang <hz...@hotmail.com>
AuthorDate: Thu Oct 18 11:23:43 2018 -0700

    KAFKA-7464; catch exceptions in "leaderEndpoint.close()" when shutting down ReplicaFetcherThread
    
    After KAFKA-6051, we close leaderEndPoint in replica fetcher thread initiateShutdown to try to preempt in-progress fetch request and accelerate repica fetcher thread shutdown. However, leaderEndpoint can throw an Exception when the replica fetcher thread is still actively fetching, which can cause ReplicaManager to fail to shutdown cleanly. This PR catches the exceptions thrown in "leaderEndpoint.close()" instead of letting it throw up in the call stack.
    
    Author: Zhanxiang (Patrick) Huang <hz...@hotmail.com>
    
    Reviewers: Manikumar Reddy <ma...@gmail.com>, Dong Lin <li...@gmail.com>
    
    Closes #5808 from hzxa21/KAFKA-7464
---
 .../src/main/scala/kafka/server/ReplicaFetcherThread.scala | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
index 2282995..6ad375c 100644
--- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
@@ -101,7 +101,19 @@ class ReplicaFetcherThread(name: String,
   override def initiateShutdown(): Boolean = {
     val justShutdown = super.initiateShutdown()
     if (justShutdown) {
-      leaderEndpoint.close()
+      // leaderEndpoint.close() can throw an exception when the replica fetcher thread is still
+      // actively fetching because the selector can close the channel while sending the request
+      // after we initiate leaderEndpoint.close() and the leaderEndpoint.close() itself may also close
+      // the channel again. When this race condition happens, an exception will be thrown.
+      // Throwing the exception to the caller may fail the ReplicaManager shutdown. It is safe to catch
+      // the exception without here causing correctness issue because we are going to shutdown the thread
+      // and will not re-use the leaderEndpoint anyway.
+      try {
+        leaderEndpoint.close()
+      } catch {
+        case t: Throwable =>
+          debug(s"Fail to close leader endpoint $leaderEndpoint after initiating replica fetcher thread shutdown", t)
+      }
     }
     justShutdown
   }