kafka： java生產者往kafka topic傳送資料傳送失敗

阿新 • • 發佈：2019-01-02

在使用java寫的kafka生產者把資料傳送給kafka的topic時，遇到傳送三次且資料無法傳送成功的Error.

2016-09-08 10:32:49,695- [INFO] Fetching metadata from broker BrokerEndPoint(0,172.17.17.98,9092) with correlation id 0 for 1 topic(s) Set(test09026)
2016-09-08 10:32:49,936- [INFO] Connected to 172.17.17.98:9092 for producing
2016-09-08 10:32:49,962- [INFO] Disconnecting from 172.17 
.17.98:9092
2016-09-08 10:32:51,056- [INFO] Connected to localhost:9092 for producing
2016-09-08 10:32:51,057- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:51,060- [WARN] Failed to send producer request with correlation id 2 to broker 0 with data for partitions [test09026,0]
java.nio.channels.ClosedChannelException 

    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply 
$mcV$sp(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:109)
    at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:108)
    at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:275)
    at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:113)
	at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:105)
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
    at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
	at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
    at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:230)
    at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
    at scala.collection.mutable.HashMap.foreach(HashMap.scala:99)
    at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
    at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:105)
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
    at kafka.producer.Producer.send(Producer.scala:78)
    at kafka.javaapi.producer.Producer.send(Producer.scala:35)
    at ProducerDemo.run(ProducerDemo.java:32)
2016-09-08 10:32:51,084- [INFO] Back off for 100 ms before retrying send. Remaining retries = 3
2016-09-08 10:32:51,186- [INFO] Fetching metadata from broker BrokerEndPoint(0,172.17.17.98,9092) with correlation id 3 for 1 topic(s) Set(test09026)
2016-09-08 10:32:51,188- [INFO] Connected to 172.17.17.98:9092 for producing
2016-09-08 10:32:51,194- [INFO] Disconnecting from 172.17.17.98:9092
2016-09-08 10:32:51,195- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:52,207- [INFO] Connected to localhost:9092 for producing
2016-09-08 10:32:52,207- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:52,207- [WARN] Failed to send producer request with correlation id 5 to broker 0 with data for partitions [test09026,0]
java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:109)
    at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:108)
    at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:275)
    at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:113)
	at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:105)
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
    at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
	at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
    at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:230)
    at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
    at scala.collection.mutable.HashMap.foreach(HashMap.scala:99)
    at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
    at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:105)
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
    at kafka.producer.Producer.send(Producer.scala:78)
    at kafka.javaapi.producer.Producer.send(Producer.scala:35)
    at ProducerDemo.run(ProducerDemo.java:32)
2016-09-08 10:32:52,210- [INFO] Back off for 100 ms before retrying send. Remaining retries = 2
2016-09-08 10:32:52,311- [INFO] Fetching metadata from broker BrokerEndPoint(0,172.17.17.98,9092) with correlation id 6 for 1 topic(s) Set(test09026)
2016-09-08 10:32:52,313- [INFO] Connected to 172.17.17.98:9092 for producing
2016-09-08 10:32:52,321- [INFO] Disconnecting from 172.17.17.98:9092
2016-09-08 10:32:52,339- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:53,346- [INFO] Connected to localhost:9092 for producing
2016-09-08 10:32:53,346- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:53,347- [WARN] Failed to send producer request with correlation id 8 to broker 0 with data for partitions [test09026,0]
java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:109)
    at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:108)
    at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:275)
    at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:113)
	at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:105)
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
    at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
	at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
    at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:230)
    at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
    at scala.collection.mutable.HashMap.foreach(HashMap.scala:99)
    at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
    at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:105)
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
    at kafka.producer.Producer.send(Producer.scala:78)
    at kafka.javaapi.producer.Producer.send(Producer.scala:35)
    at ProducerDemo.run(ProducerDemo.java:32)
2016-09-08 10:32:53,348- [INFO] Back off for 100 ms before retrying send. Remaining retries = 1
2016-09-08 10:32:53,450- [INFO] Fetching metadata from broker BrokerEndPoint(0,172.17.17.98,9092) with correlation id 9 for 1 topic(s) Set(test09026)
2016-09-08 10:32:53,453- [INFO] Connected to 172.17.17.98:9092 for producing
2016-09-08 10:32:53,456- [INFO] Disconnecting from 172.17.17.98:9092
2016-09-08 10:32:53,457- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:54,462- [INFO] Connected to localhost:9092 for producing
2016-09-08 10:32:54,462- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:54,466- [WARN] Failed to send producer request with correlation id 11 to broker 0 with data for partitions [test09026,0]
java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:110)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:109)
    at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
	at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:109)
    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:108)
    at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:275)
    at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:113)
	at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:105)
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
    at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
	at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
    at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:230)
    at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
    at scala.collection.mutable.HashMap.foreach(HashMap.scala:99)
    at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
    at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:105)
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
    at kafka.producer.Producer.send(Producer.scala:78)
    at kafka.javaapi.producer.Producer.send(Producer.scala:35)
    at ProducerDemo.run(ProducerDemo.java:32)
2016-09-08 10:32:54,468- [INFO] Back off for 100 ms before retrying send. Remaining retries = 0
2016-09-08 10:32:54,582- [INFO] Fetching metadata from broker BrokerEndPoint(0,172.17.17.98,9092) with correlation id 12 for 1 topic(s) Set(test09026)
2016-09-08 10:32:54,585- [INFO] Connected to 172.17.17.98:9092 for producing
2016-09-08 10:32:54,588- [INFO] Disconnecting from 172.17.17.98:9092
2016-09-08 10:32:54,588- [INFO] Disconnecting from localhost:9092
2016-09-08 10:32:54,591- [ERROR] Failed to send requests for topics test09026 with correlation ids in [0,12]
Exception in thread "Thread-0" kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:96)
    at kafka.producer.Producer.send(Producer.scala:78)
    at kafka.javaapi.producer.Producer.send(Producer.scala:35)
    at ProducerDemo.run(ProducerDemo.java:32)

我是從window伺服器往linux伺服器傳送內容的。

解決方案：
1.在windows的host檔案中加上linux伺服器ip和hostname的對映。
2.在server.properteies中修改如下配置

#advertised.listeners=PLAINTEXT://your.host.name:9092

修改成

advertised.listeners=PLAINTEXT://172.16.128.208:9092

kafka： java生產者往kafka topic傳送資料傳送失敗

在使用java寫的kafka生產者把資料傳送給kafka的topic時，遇到傳送三次且資料無法傳送成功的Error. 2016-09-08 10:32:49,695- [INFO] Fetching metadata from broker BrokerE

Kafka：無丟失提取kafka的值，詳解kafka的消費過程

目錄： 1、需求 2、程式碼步鄹 3、程式碼展現 4、pom.xml檔案 5、結果展現 ——————————————————————————————————– 1、需求前提：將org.apache.spark.streaming.kafka.KafkaCluster這個類抽出來變成Kafka

Java：java爬蟲獲取動態網頁的資料

說明：只是分享一種解決方案，程式碼以及部分截圖不方便貼出，請諒解！前段時間一直在研究爬蟲，抓取網路上的特定的資料，如果只是靜態網頁就是再簡單不過了，直接使用Jsoup ： Document doc = Jsoup.connect(url).timeout(2000).

小例子：java利用poi讀取excel中資料並匯入資料庫

問題描述：資料夾下有若干excel檔案,檔名為10.教育局.xls 11.衛生院.xls ................有很多；中間的漢字為單位名稱，需要匯入資料庫，每個單位名稱要有一個單位id匹配；每個excel中有若干個sheet頁的名字,每個名字即為科

Java——定時請求後端介面資料傳送RabbitMQ訊息佇列到指定MQ伺服器

RabbitMQ入門參考部落格：http://m.blog.csdn.net/article/details?id=50487028（內含介紹、環境配置及基本實現說明） RabbitMQ官方文件參考，傳送Topic exchange佇列：http://www.rabbitm

基礎篇：JAVA.Stream函式，優雅的資料流操作

# 前言平時操作集合資料，我們一般都是for或者iterator去遍歷，不是很好看。java提供了Stream的概念，它可以讓我們把集合資料當做一個個元素在處理，並且提供多執行緒模式 - 流的建立 - 流的各種資料操作 - 流的終止操作 - 流的聚合處理 - 併發流和CompletableFuture的配

Android GET,POST向伺服器端傳送資料(傳送)

//目錄結構 //strings.xml字元常量檔案 <?xml version="1.0" encoding="utf-8"?> <resources> <string name="title">通過Get和Pos

kafka原始碼解析之十六生產者流程(客戶端如何向topic傳送資料)

客戶端向topic傳送資料分為兩種方式：1.非同步，2同步。其配置為producer.type，如果為sync，則是同步傳送；如果為async，則是非同步傳送。客戶端程式碼如下： import kafka.javaapi.producer.Producer; import

Kafka：ZK+Kafka+Spark Streaming集群環境搭建（二十五）Structured Streaming：同一個topic中包含一組數據的多個部分，按照key它們拼接為一條記錄（以及遇到的問題）。

eas array 記錄 splay span ack timestamp b- each 需求：目前kafka的topic上有一批數據，這些數據被分配到9個不同的partition中（就是發布時key:{m1,m2,m3,m4...m9},value:{records

kafka-0.10.0.1版本在java客戶端傳送資料接收不到

錯誤日誌： org.apache.kafka.common.errors.TimeoutException: Batch containing 100 record kafka版本：0.10.0.1 zookeeper版本: 3.4.5 產生的現象: 編寫好java

快手Java四面：設計模式+紅黑樹+Java鎖+Redis+Kafka等

一面(一個小時十分鐘) 1.自我介紹 2.說說B+樹和B樹的區別，優缺點等？ 3聊聊Spring，主要IOC等等 4多執行緒JUC包下的一些常見的類，比如CountDownLatch、Semaphore等 5.鎖的概念，鎖相關的關鍵字，volatile，synchronized。

java向kafka批量均勻傳送資料的方法

RT。import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.P

kafka單條訊息過大導致生產者程式傳送到broker失敗

今天碰到一個問題，kafka生產者罷工停止生產了，而且生產者的記憶體急劇升高，導致程式幾次重啟。檢視日誌，才發現生產者程式爆出異常kafka.common.MessageSizeTooLargeExc

kafka消費者、生產者，Java實現

參考：https://www.cnblogs.com/zlslch/p/5966004.html1、KafkaProducerOps.java//執行成功 import org.apache.kafka.clients.producer.KafkaProducer; impo

使用 kafka 的java客戶端進行訊息的傳送與接收通訊操作

kafka的傳送端： package com.zwz.test; import kafka.Kafka; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.cli

四面快手歸來，分享Java真題及面經：策略模式+紅黑樹+Java鎖+Redis+Kafka等分散式

看真題，瞭解差距，明確學習方向與目標。看面經，提前準備，事半功倍。 ** 一面(一個小時十分鐘) ** 1.自我介紹 2.說說B+樹和B樹的區別，優缺點等？ 3聊聊Spring，主要IOC等等 4多執行緒JUC包下的一些常見的類，比如CountDownLatch、Se

kafka傳送訊息出現的問題KafKa error java.nio.channels.UnresolvedAddressException

出現以下的錯 124 [main] INFO kafka.utils.VerifiableProperties - Verifying properties 157 [main] INFO kafka.utils.VerifiableProperties - Pr

深入學習Kafka：Topic的刪除過程分析

要刪除Topic，需要執行下面命令： .\kafka-topics.bat --delete --zookeeper localhost:2181 --topic test 這裡假設zookeeper地址為localhost，要刪除的topic是te

kafka啟動時報錯：java.net.UnknownHostException: Name or service not known

今天在用下面的指令啟動kafa的時候報錯了： # bin/kafka-server-start.sh config/server.properties & 錯誤資訊如下： [2018-04-11 16:27:31,185] ERROR [KafkaServer

Streaming消費kafka報錯：java.lang.NoClassDefFoundError: net/jpountz/util/SafeUtils

1.問題描述 Streaming消費kafka訊息，報以下錯誤： 18/11/22 18:14:55 ERROR Executor: Exception in task 0.0 in stage 14.0 (TID 9) java.lang.NoClassDefFound

kafka： java生產者往kafka topic傳送資料傳送失敗

相關推薦