[jira] [Created] (KAFKA-904) ClientUtils throws SocketTimeOut in client side at high load

["Rajesh Balamohan (JIRA)" <ji...@apache.org>]

Rajesh Balamohan created KAFKA-904:
--------------------------------------

             Summary: ClientUtils throws SocketTimeOut in client side at high load
                 Key: KAFKA-904
                 URL: https://issues.apache.org/jira/browse/KAFKA-904
             Project: Kafka
          Issue Type: Improvement
         Environment: Redhat linux with 3 physical nodes.
            Reporter: Rajesh Balamohan


When performance benchmarking under heavy load, we encountered few "SocketTimeOut" exception in client side (producer pushing then message to broker).

Looking at the stacktrace, the issue stemmed from core/src/main/scala/kafka/client/ClientUtils.scala

We provide 3 brokers to the list. It appears that Kafka always hits the first broker and if its down, it would not remove the id from the broker list. This could cause delays in message processing and also bring down performance.

Can we have different strategies like round-robin, weighted round robin etc to decide on which broker to connect to?. Also, if a broker is down it should be removed from the broker list.

 /**
   * Used by the producer to send a metadata request since it has access to the ProducerConfig
   * @param topics The topics for which the metadata needs to be fetched
   * @param brokers The brokers in the cluster as configured on the producer through broker.list
   * @param producerConfig The producer's config
   * @return topic metadata response
   */
  def fetchTopicMetadata(topics: Set[String], brokers: Seq[Broker], producerConfig: ProducerConfig, correlationId: Int): TopicMetadataResponse = {
    var fetchMetaDataSucceeded: Boolean = false
    var i: Int = 0
    val topicMetadataRequest = new TopicMetadataRequest(TopicMetadataRequest.CurrentVersion, correlationId, producerConfig.clientId, topics.toSeq)
    var topicMetadataResponse: TopicMetadataResponse = null
    var t: Throwable = null
    while(i < brokers.size && !fetchMetaDataSucceeded) {
      val producer: SyncProducer = ProducerPool.createSyncProducer(producerConfig, brokers(i))
      info("Fetching metadata with correlation id %d for %d topic(s) %s".format(correlationId, topics.size, topics))
      try {
        topicMetadataResponse = producer.send(topicMetadataRequest)
        fetchMetaDataSucceeded = true
      }
      catch {
        case e =>
          warn("Fetching topic metadata with correlation id %d for topics [%s] from broker [%s] failed"
            .format(correlationId, topics, brokers(i).toString), e)
          t = e
      } finally {
        i = i + 1
        producer.close()
      }
    }
    if(!fetchMetaDataSucceeded){
      throw new KafkaException("fetching topic metadata for topics [%s] from broker [%s] failed".format(topics, brokers), t)
    } else {
      debug("Successfully fetched metadata for %d topic(s) %s".format(topics.size, topics))
    }
    return topicMetadataResponse
  }



--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira