Uploaded image for project: 'Kafka'
  1. Kafka
  2. KAFKA-8066

ReplicaFetcherThread fails to startup because of failing to register the metric.

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Resolved
    • Priority: Major
    • Resolution: Fixed
    • Affects Version/s: 1.1.0, 1.1.1, 1.1.2, 2.0.0, 2.0.1, 2.1.0, 2.2.0, 2.1.1, 2.0.2, 2.1.2
    • Fix Version/s: 2.0.2, 2.1.2, 2.2.1
    • Component/s: None
    • Labels:
      None

      Description

      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, the selector may fail to close the channel and throw an Exception when the replica fetcher thread is still actively fetching. In this case, the sensor will not be cleaned up. Basically, if `close(id)` throws an exception in `Selector.close()`, then `sensors.close()` will not be called and thus the sensors will not get unregistered (See codes below).

          public void close() {
              List<String> connections = new ArrayList<>(channels.keySet());
              for (String id : connections)
                  close(id);
              try {
                  this.nioSelector.close();
              } catch (IOException | SecurityException e) {
                  log.error("Exception closing nioSelector:", e);
              }
              sensors.close();
              channelBuilder.close();
          }
      

      If this happen, when the broker want to start up the ReplicaFetcherThread with the same fetch id to the same destination broker again (e.g. due to leadership changes or new partitions get created), the ReplicaFetcherThread will fail to start up because the selector will throw an IllegalArgumentException if the metric with the same name already exists:

      2019/02/27 10:24:26.938 ERROR [KafkaApis] [kafka-request-handler-6] [kafka-server] [] [KafkaApi-38031] Error when handling request {}
      java.lang.IllegalArgumentException: A metric named 'MetricName [name=connection-count, group=replica-fetcher-metrics, description=The current number of active connections., tags={broker-id=29712, fetcher-id=3}]' already exists, can't register another one.
              at org.apache.kafka.common.metrics.Metrics.registerMetric(Metrics.java:559) ~[kafka-clients-2.0.0.66.jar:?]
              at org.apache.kafka.common.metrics.Metrics.addMetric(Metrics.java:502) ~[kafka-clients-2.0.0.66.jar:?]
              at org.apache.kafka.common.metrics.Metrics.addMetric(Metrics.java:485) ~[kafka-clients-2.0.0.66.jar:?]
              at org.apache.kafka.common.metrics.Metrics.addMetric(Metrics.java:470) ~[kafka-clients-2.0.0.66.jar:?]
              at org.apache.kafka.common.network.Selector$SelectorMetrics.<init>(Selector.java:963) ~[kafka-clients-2.0.0.66.jar:?]
              at org.apache.kafka.common.network.Selector.<init>(Selector.java:170) ~[kafka-clients-2.0.0.66.jar:?]
              at org.apache.kafka.common.network.Selector.<init>(Selector.java:188) ~[kafka-clients-2.0.0.66.jar:?]
              at kafka.server.ReplicaFetcherBlockingSend.<init>(ReplicaFetcherBlockingSend.scala:61) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.ReplicaFetcherThread$$anonfun$1.apply(ReplicaFetcherThread.scala:68) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.ReplicaFetcherThread$$anonfun$1.apply(ReplicaFetcherThread.scala:68) ~[kafka_2.11-2.0.0.66.jar:?]
              at scala.Option.getOrElse(Option.scala:121) ~[scala-library-2.11.12.jar:?]
              at kafka.server.ReplicaFetcherThread.<init>(ReplicaFetcherThread.scala:67) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.ReplicaFetcherManager.createFetcherThread(ReplicaFetcherManager.scala:32) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.AbstractFetcherManager.kafka$server$AbstractFetcherManager$$addAndStartFetcherThread$1(AbstractFetcherManager.scala:132) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.AbstractFetcherManager$$anonfun$addFetcherForPartitions$2.apply(AbstractFetcherManager.scala:146) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.AbstractFetcherManager$$anonfun$addFetcherForPartitions$2.apply(AbstractFetcherManager.scala:137) ~[kafka_2.11-2.0.0.66.jar:?]
              at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733) ~[scala-library-2.11.12.jar:?]
              at scala.collection.immutable.Map$Map1.foreach(Map.scala:116) ~[scala-library-2.11.12.jar:?]
              at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732) ~[scala-library-2.11.12.jar:?]
              at kafka.server.AbstractFetcherManager.addFetcherForPartitions(AbstractFetcherManager.scala:137) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.ReplicaManager.makeFollowers(ReplicaManager.scala:1333) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.ReplicaManager.becomeLeaderOrFollower(ReplicaManager.scala:1107) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.KafkaApis.handleLeaderAndIsrRequest(KafkaApis.scala:194) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.KafkaApis.handle(KafkaApis.scala:110) ~[kafka_2.11-2.0.0.66.jar:?]
              at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69) ~[kafka_2.11-2.0.0.66.jar:?]
              at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
      
      

      The fix should be adding a try-finally block for selector.close() to make sure sensors.close() will be called even an exception is thrown.

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                hzxa21 Zhanxiang (Patrick) Huang
                Reporter:
                hzxa21 Zhanxiang (Patrick) Huang
              • Votes:
                0 Vote for this issue
                Watchers:
                3 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: