kafka 生產發送消息失敗無響應或者Error while fetching metadata with correlation id該怎么辦

kafka 生產發送消息失敗無響應或者Error while fetching metadata with correlation id該怎么辦，很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學習下，希望你能有所收獲。

今天在使用代碼編寫kafka 生產者發送消息的時候，因為我的手誤出現的搞笑的事情。

同樣的代碼和kafka 在不久前執行過，是沒有問題的。 代碼如下

package streaming.utils

import java.util
import java.util.{Date, Properties, UUID}

import com.alibaba.fastjson.JSONObject
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.util.Random

/**
  * Author: Michael PK   QQ: 1990218038
  *
  * Kafka數據生產者
  */
object ProducerApp {

  def main(args: Array[String]): Unit = {

    val props = new Properties
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("bootstrap.servers", ParamsConf.brokers)
    props.put("request.required.acks","1")

    val topic = ParamsConf.topic
    val producer = new KafkaProducer[String,String](props)

    val random = new Random()
    val dateFormat = FastDateFormat.getInstance("yyyyMMddHHmmss")

    for(i <- 1 to 100){
      val time = dateFormat.format(new Date())+""
      val userid = random.nextInt(1000)+""
      val courseid = random.nextInt(500)+""
      val fee = random.nextInt(400)+""
      val result = Array("0","1") // 0未成功支付，1成功支付
      val flag = result(random.nextInt(2))
      var orderid = UUID.randomUUID().toString

      val map = new util.HashMap[String, Object]()
      map.put("time", time)
      map.put("userid",userid)
      map.put("courseid",courseid)
      map.put("fee", fee)
      map.put("flag", flag)
      map.put("orderid",orderid)

      val json = new JSONObject(map)

      producer.send(new ProducerRecord[String,String](topic(0),json.toJSONString))
    }


    println("PK Kafka生產者生產數據完畢...")
  }

}

代碼很簡單。只是用來模擬生產數據而已。

一直以來的使用的 都是 2.0 版本的 kafka client

<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.0.0</version>
        </dependency>

但是今天執行的上面的代碼的時候。 就 不能發生消息了，也沒有錯誤的提示。 程序也沒有關閉。

通過 debug 發現卡在  doSend 代碼里面了

 private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
        TopicPartition tp = null;
        try {
            throwIfProducerClosed();
            // first make sure the metadata for the topic is available
            ClusterAndWaitTime clusterAndWaitTime;
            try {
                clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
            } catch (KafkaException e) {
                if (metadata.isClosed())
                    throw new KafkaException("Producer closed while send in progress", e);
                throw e;
            }

雖然它拋出了異常，但是 不能進入 

if (metadata.isClosed()) 邏輯里面

外層并沒有捕獲它的異常。通過debug 這個 異常 e 是    Failed to update metadata after 60000 ms.

考慮到它的版本也服務器版本不一樣 就試著 減低版本看看

 <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>1.1.0</version>
        </dependency>

執行的時候不停的提示：因為進入了死循環中

lientId=producer-1] Error while fetching metadata with correlation id 

并且測試過了 ： telnet 192.168.0.205 9092  是沒有問題的

在 服務器本地上面使用 命令生產消費消息是可以的。

這樣就奇怪了。 于是我換成了 另一個 主題進行測試發現是沒有問題的。。。

同時注意到了原來是 我寫的 主題名稱 后面帶有空格

低級錯誤啊！但是后面的空格真的是空格？我自己測試手敲空格，經過測試沒有問題的。也就是如果是 主題后面有空格是可以的。

那就是主題名稱 后面帶上了什么不可見內容。 我想起來了，我是通過復制這個主題名稱 的，估計復制多了什么其他內容。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。