kafka 生產(chǎn)發(fā)送消息失敗無響應(yīng)或者Error while fetching metadata with correlation id該怎么辦,很多新手對此不是很清楚,為了幫助大家解決這個難題,下面小編將為大家詳細(xì)講解,有這方面需求的人可以來學(xué)習(xí)下,希望你能有所收獲。
我們提供的服務(wù)有:網(wǎng)站設(shè)計、網(wǎng)站建設(shè)、微信公眾號開發(fā)、網(wǎng)站優(yōu)化、網(wǎng)站認(rèn)證、鳳陽ssl等。為1000+企事業(yè)單位解決了網(wǎng)站和推廣的問題。提供周到的售前咨詢和貼心的售后服務(wù),是有科學(xué)管理、有技術(shù)的鳳陽網(wǎng)站制作公司
今天在使用代碼編寫kafka 生產(chǎn)者發(fā)送消息的時候,因為我的手誤出現(xiàn)的搞笑的事情。
同樣的代碼和kafka 在不久前執(zhí)行過,是沒有問題的。 代碼如下
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數(shù)據(jù)生產(chǎn)者 */ 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生產(chǎn)者生產(chǎn)數(shù)據(jù)完畢...") } }
代碼很簡單。只是用來模擬生產(chǎn)數(shù)據(jù)而已。
一直以來的使用的 都是 2.0 版本的 kafka client
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.0.0</version> </dependency>
但是今天執(zhí)行的上面的代碼的時候。 就 不能發(fā)生消息了,也沒有錯誤的提示。 程序也沒有關(guān)閉。
通過 debug 發(fā)現(xiàn)卡在 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.
考慮到它的版本也服務(wù)器版本不一樣 就試著 減低版本看看
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>1.1.0</version> </dependency>
執(zhí)行的時候不停的提示:因為進入了死循環(huán)中
lientId=producer-1] Error while fetching metadata with correlation id
并且測試過了 : telnet 192.168.0.205 9092 是沒有問題的
在 服務(wù)器本地上面使用 命令生產(chǎn)消費消息是可以的。
這樣就奇怪了。 于是我換成了 另一個 主題進行測試發(fā)現(xiàn)是沒有問題的。。。
同時注意到了原來是 我寫的 主題名稱 后面帶有空格
低級錯誤啊!但是后面的空格真的是空格?我自己測試手敲空格,經(jīng)過測試沒有問題的。也就是如果是 主題后面有空格是可以的。
那就是主題名稱 后面帶上了什么不可見內(nèi)容。 我想起來了,我是通過復(fù)制這個主題名稱 的,估計復(fù)制多了什么其他內(nèi)容。
看完上述內(nèi)容是否對您有幫助呢?如果還想對相關(guān)知識有進一步的了解或閱讀更多相關(guān)文章,請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道,感謝您對創(chuàng)新互聯(lián)的支持。
新聞名稱:kafka生產(chǎn)發(fā)送消息失敗無響應(yīng)或者Errorwhilefetchingmetadatawithcorrelationid該怎么辦
轉(zhuǎn)載來源:http://aaarwkj.com/article44/jjgjhe.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供做網(wǎng)站、云服務(wù)器、軟件開發(fā)、用戶體驗、小程序開發(fā)、網(wǎng)站收錄
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)