[kafka] branch 2.1 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.1
in repository https://gitbox.apache.org/repos/asf/kafka.git


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

commit e559bfdd1cebc8c26754d59d04d162fa3813806e
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
    
    (cherry picked from commit 928f45f61f793f0b2d7df11ff92463d7edda60d6)
    Signed-off-by: Dong Lin <li...@gmail.com>
---
 .../scala/kafka/server/ReplicaFetcherThread.scala  | 14 ++++++-
 .../kafka/server/ReplicaFetcherThreadTest.scala    | 43 +++++++++++++++++++++-
 2 files changed, 54 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
index aeeaf29..71a4b95 100644
--- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
@@ -115,7 +115,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
   }
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
index 4d54c81..c9d9b96 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
@@ -24,16 +24,16 @@ import kafka.cluster.Partition
 import kafka.server.QuotaFactory.UnboundedQuota
 import kafka.server.epoch.LeaderEpochFileCache
 import kafka.server.epoch.util.ReplicaFetcherMockBlockingSend
-import kafka.utils.TestUtils
+import kafka.utils.{LogCaptureAppender, TestUtils}
 import org.apache.kafka.clients.ClientResponse
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.internals.PartitionStates
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.protocol.Errors._
 import org.apache.kafka.common.requests.{EpochEndOffset, OffsetsForLeaderEpochRequest}
 import org.apache.kafka.common.requests.EpochEndOffset._
 import org.apache.kafka.common.utils.SystemTime
+import org.apache.log4j.Level
 import org.easymock.EasyMock._
 import org.easymock.{Capture, CaptureType, IAnswer}
 import org.junit.Assert._
@@ -793,6 +793,45 @@ class ReplicaFetcherThreadTest {
     assertEquals(49, truncateToCapture.getValue)
   }
 
+  @Test
+  def shouldCatchExceptionFromBlockingSendWhenShuttingDownReplicaFetcherThread(): Unit = {
+    val props = TestUtils.createBrokerConfig(1, "localhost:1234")
+    val config = KafkaConfig.fromProps(props)
+    val mockBlockingSend = createMock(classOf[BlockingSend])
+
+    expect(mockBlockingSend.close()).andThrow(new IllegalArgumentException()).once()
+    replay(mockBlockingSend)
+
+    val thread = new ReplicaFetcherThread(
+      name = "bob",
+      fetcherId = 0,
+      sourceBroker = brokerEndPoint,
+      brokerConfig = config,
+      replicaMgr = null,
+      metrics =  new Metrics(),
+      time = new SystemTime(),
+      quota = null,
+      leaderEndpointBlockingSend = Some(mockBlockingSend))
+
+    val previousLevel = LogCaptureAppender.setClassLoggerLevel(thread.getClass, Level.DEBUG)
+    val logCaptureAppender = LogCaptureAppender.createAndRegister()
+
+    try {
+      thread.initiateShutdown()
+
+      val event = logCaptureAppender.getMessages.find(e => e.getLevel == Level.DEBUG
+        && e.getRenderedMessage.contains(s"Fail to close leader endpoint $mockBlockingSend after initiating replica fetcher thread shutdown")
+        && e.getThrowableInformation != null
+        && e.getThrowableInformation.getThrowable.getClass.getName.equals(new IllegalArgumentException().getClass.getName))
+      assertTrue(event.isDefined)
+
+      verify(mockBlockingSend)
+    } finally {
+      LogCaptureAppender.unregister(logCaptureAppender)
+      LogCaptureAppender.setClassLoggerLevel(thread.getClass, previousLevel)
+    }
+  }
+
   def stub(replica: Replica, partition: Partition, replicaManager: ReplicaManager) = {
     expect(replicaManager.localReplica(t1p0)).andReturn(Some(replica)).anyTimes()
     expect(replicaManager.localReplicaOrException(t1p0)).andReturn(replica).anyTimes()