[LI-HOTFIX] Reduce lock retention and improve broker shutdown time:

TICKET = [KAFKA-8667, KAFKA-8668]
LI_DESCRIPTION =
- Avoid acquiring partitionMap lock in shutdownIdleFetcherThread
- Avoid appending to the time index during shutdown if the time index has not yet be initialized

RB=1431408
BUG=LIKAFKA-19361
G=Kafka-Code-Reviews
R=jkoshy,jonlee
A=jkoshy,jonlee

EXIT_CRITERIA = TICKET [KAFKA-8667, KAFKA-8668]

core/src/main/scala/kafka/log/LogSegment.scala

@@ -585,7 +585,9 @@ class LogSegment private[log] (val log: FileRecords,
    * Close this log segment
    */
   def close(): Unit = {
-    CoreUtils.swallow(timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, skipFullCheck = true), this)
+    if (_maxTimestampSoFar.nonEmpty || _offsetOfMaxTimestampSoFar.nonEmpty) {
+      CoreUtils.swallow(timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, skipFullCheck = true), this)
+    }
     CoreUtils.swallow(offsetIndex.close(), this)
     CoreUtils.swallow(timeIndex.close(), this)
     CoreUtils.swallow(log.close(), this)

core/src/main/scala/kafka/server/AbstractFetcherManager.scala

@@ -196,7 +196,7 @@ abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: Stri
     lock synchronized {
       val keysToBeRemoved = new mutable.HashSet[BrokerIdAndFetcherId]
       for ((key, fetcher) <- fetcherThreadMap) {
-        if (fetcher.partitionCount <= 0) {
+        if (fetcher.idle) {
           fetcher.shutdown()
           keysToBeRemoved += key
         }

core/src/main/scala/kafka/server/AbstractFetcherThread.scala

@@ -68,6 +68,8 @@ abstract class AbstractFetcherThread(name: String,
   val fetcherStats = new FetcherStats(metricId)
   val fetcherLagStats = new FetcherLagStats(metricId)
 
+  @volatile var idle = false
+
   /* callbacks to be defined in subclass */
 
   // process fetched data
@@ -651,6 +653,7 @@ abstract class AbstractFetcherThread(name: String,
         partitionStates.remove(topicPartition)
         fetcherLagStats.unregister(topicPartition)
       }
+      idle = partitionStates.size() <= 0
     } finally partitionMapLock.unlock()
   }