WARN [ReplicaFetcherThread-1-1011], Error in fetch kafka.server.ReplicaFetcherThread$FetchRequest@5dad549c

[Tony Liu]

​Hi Experts,

is there anyone run into this connection error ?

[2016-12-17 20:13:32,728] WARN  [ReplicaFetcherThread-1-1011], Error
in fetch kafka.server.ReplicaFetcherThread$FetchRequest@5dad549c
(kafka.server.ReplicaFetcherThread)java.io.IOException: Connection to
10.137.126.113:9092 (id: 1011 rack: null) failed
        at 
kafka.utils.NetworkClientBlockingOps$.awaitReady$1(NetworkClientBlockingOps.scala:83)
        at 
kafka.utils.NetworkClientBlockingOps$.blockingReady$extension(NetworkClientBlockingOps.scala:93)
        at 
kafka.server.ReplicaFetcherThread.sendRequest(ReplicaFetcherThread.scala:248)
        at 
kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:238)
        at 
kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:42)
        at 
kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:118)
        at 
kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:103)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)



by the way, I check the source code, which said there are lots of reasons
causing that error, so I am not clear about what the 'lots of reasons` are,
so that I put effort to do further addressing.

/**
 * Invokes `client.send` followed by 1 or more `client.poll`
invocations until a response is received or a
 * disconnection happens (which can happen for a number of reasons
including a request timeout).
 *
 * In case of a disconnection, an `IOException` is thrown.
 *
 * This method is useful for implementing blocking behaviour on top of
the non-blocking `NetworkClient`, use it with
 * care.
 */
def blockingSendAndReceive(request: ClientRequest)(implicit time:
JTime): ClientResponse = {
  client.send(request, time.milliseconds())

  pollContinuously { responses =>
    val response = responses.find { response =>
      response.request.request.header.correlationId ==
request.request.header.correlationId
    }
    response.foreach { r =>
      if (r.wasDisconnected) {
        val destination = request.request.destination
        throw new IOException(s"Connection to $destination was
disconnected before the response was read")
      }
    }
    response
  }

}