在HDP Sandbox 2.6.5中从localhost生成Kafka不起作用
在HDP Sandbox 2.6.5中从localhost生成Kafka不起作用
提问于 2018-09-12 20:40:59
我把Kafka客户端生产者写成：
public class BasicProducerExample {
public static void main(String[] args){
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//props.put(ProducerConfig.
props.put("batch.size","16384");// maximum size of message
Producer<String, String> producer = new KafkaProducer<String, String>(props);
TestCallback callback = new TestCallback();
Random rnd = new Random();
for (long i = 0; i < 2 ; i++) {
//ProducerRecord<String, String> data = new ProducerRecord<String, String>("dke", "key-" + i, "message-"+i );
//Topci and Message
ProducerRecord<String, String> data = new ProducerRecord<String, String>("dke", ""+i);
producer.send(data, callback);
}
producer.close();
}
private static class TestCallback implements Callback {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
System.out.println("Error while producing message to topic :" + recordMetadata);
e.printStackTrace();
} else {
String message = String.format("sent message to topic:%s partition:%s offset:%s", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
System.out.println(message);
}
}
}
}输出:向主题生成消息时出错:空org.apache.kafka.common.errors.TimeoutException:无法在60000毫秒后更新元数据。
注意: Broker端口: localhost:6667正在工作。
回答 2
Stack Overflow用户
发布于 2018-09-12 21:32:49
在BOOTSTRAP_SERVERS_CONFIG的属性中，尝试将端口号更改为6667。
谢谢。
--海伦
Stack Overflow用户
发布于 2019-02-11 18:19:26
我在Hortonworks (HDP 2.X版本)安装上使用Apache Kafka。遇到的错误消息表示Kafka producer无法将数据推送到段日志文件。在命令行控制台中，这意味着两件事：
- 您正在为代理使用不正确的端口
- 您的监听器配置在server.properties中不起作用
如果您在通过scala api编写时遇到错误消息，请另外使用telnet <cluster-host> <broker-port>检查与kafka集群的连接
注意：如果您正在使用scala api创建主题，那么代理需要一段时间才能知道新创建的主题。因此，在主题创建之后，生产者可能会立即失败，并返回错误Failed to update metadata after 60000 ms.
为了解决这个问题，我做了以下检查：
我通过Ambari检查的第一个区别是，Kafka代理监听HDP2.x上的端口6667 (apache kafka使用9092)。
listeners=PLAINTEXT://localhost:6667接下来，使用ip而不是localhost。我处死了netstat -na | grep 6667
tcp 0 0 192.30.1.5:6667 0.0.0.0:* LISTEN
tcp 1 0 192.30.1.5:52242 192.30.1.5:6667 CLOSE_WAIT
tcp 0 0 192.30.1.5:54454 192.30.1.5:6667 TIME_WAIT因此，我将生产者调用修改为使用IP而不是localhost：
./kafka-console-producer.sh --broker-list 192.30.1.5:6667 --topic rdl_test_2要监视是否有新记录正在写入，请监视/kafka-logs文件夹。
cd /kafka-logs/<topic name>/
ls -lart
-rw-r--r--. 1 kafka hadoop 0 Feb 10 07:24 00000000000000000000.log
-rw-r--r--. 1 kafka hadoop 10485756 Feb 10 07:24 00000000000000000000.timeindex
-rw-r--r--. 1 kafka hadoop 10485760 Feb 10 07:24 00000000000000000000.index一旦生产者成功写入，段日志文件00000000000000000000.log的大小就会增加。
大小见下图：
-rw-r--r--. 1 kafka hadoop 10485760 Feb 10 07:24 00000000000000000000.index
-rw-r--r--. 1 kafka hadoop **45** Feb 10 09:16 00000000000000000000.log
-rw-r--r--. 1 kafka hadoop 10485756 Feb 10 07:24 00000000000000000000.timeindex此时，您可以运行consumer-console.sh：
./kafka-console-consumer.sh --bootstrap-server 192.30.1.5:6667 --topic rdl_test_2 --from-beginning
response is hello world完成此步骤后，如果您希望通过Scala API生成消息，请更改listeners值(从本地主机更改为公网IP)，然后通过Ambari重启Kafka brokers：
listeners=PLAINTEXT://192.30.1.5:6667 示例生成器如下所示：
package com.scalakafka.sample
import java.util.Properties
import java.util.concurrent.TimeUnit
import org.apache.kafka.clients.producer.{ProducerRecord, KafkaProducer}
import org.apache.kafka.common.serialization.{StringSerializer, StringDeserializer}
class SampleKafkaProducer {
case class KafkaProducerConfigs(brokerList: String = "192.30.1.5:6667") {
val properties = new Properties()
val batchsize :java.lang.Integer = 1
properties.put("bootstrap.servers", brokerList)
properties.put("key.serializer", classOf[StringSerializer])
properties.put("value.serializer", classOf[StringSerializer])
// properties.put("serializer.class", classOf[StringDeserializer])
properties.put("batch.size", batchsize)
// properties.put("linger.ms", 1)
// properties.put("buffer.memory", 33554432)
}
val producer = new KafkaProducer[String, String](KafkaProducerConfigs().properties)
def produce(topic: String, messages: Iterable[String]): Unit = {
messages.foreach { m =>
println(s"Sending $topic and message is $m")
val result = producer.send(new ProducerRecord(topic, m)).get()
println(s"the write status is ${result}")
}
producer.flush()
producer.close(10L, TimeUnit.MILLISECONDS)
}
}希望这对某些人有帮助。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/52295443
复制相关文章
相似问题
腾讯云开发者
