Apache Kafka示例错误:3次尝试后无法发送消息

Question

我正在运行其网站中提到的这个kafka生产者示例

代码:

public class TestProducer {

    public static void main(String[] args) {
        long events = Long.parseLong(args[0]);
        Random rnd = new Random();
        Properties props = new Properties();
        props.put("metadata.broker.list", "host.broker-1:9093, host.broker-2:9093, host.broker-3:9095");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "test.app.SimplePartitioner");
        props.put("request.required.acks", "1");
        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);
        for (long nEvents = 0; nEvents < events; nEvents++) { 
               long runtime = new Date().getTime();  
               String ip = "192.168.2." + rnd.nextInt(255); 
               String msg = runtime + ",www.example.com," + ip; 
               KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
               producer.send(data);
        }
        producer.close();
    }
}

public class SimplePartitioner implements Partitioner{
    public SimplePartitioner (VerifiableProperties props) {

    }

    public int partition(Object key, int a_numPartitions) {
        int partition = 0;
        String stringKey = (String) key;
        int offset = stringKey.lastIndexOf('.');
        if (offset > 0) {
           partition = Integer.parseInt( stringKey.substring(offset+1)) % a_numPartitions;
        }
       return partition;
  }

}

更多细节:

我在主机(调用生产者)上运行此应用程序,该主机远离主机代理[1-3]

• 我可以从生产者主机ping和ssh代理主机.

• 在server.properties中提供了advertised.host.name(它们分别在代理中命名为server [1-3] .properties

属性:

broker.id=1
port=9093
host.name=host.broker.internal.name
advertised.host.name=host-broker1
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/1/kafka-logs-1,/data/2/kafka-logs-2
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
zookeeper.connection.timeout.ms=6000

有关如何解决此错误的任何想法？

Answer 1

运行Kafka生产者时遇到这些错误:

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

找到了解决方案:

在我的Mac机箱上,在我下载了scala-2.10和之后kafka_2.10-0.8.1,在kafka_2.10-0.8.1目录中,当我启动zookeeper,kafka服务器并创建一个测试主题时,一切都很好.然后我需要为测试主题启动一个生产者.但是有一个错误:

yhuangMac:kafka_2.10-0.8.1 yhuang$ ./bin/kafka-console-producer.sh –broker-list localhost:9092 –topic test
SLF4J: Failed to load class “org.slf4j.impl.StaticLoggerBinder”.
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

原因是在kafka libs目录中,kafka发布的zip文件只包含了slf4j-api的jar文件,他们错过了一个jar文件:slf4j-nop.jar,所以我们必须去http://www.slf4j.组织,下载slf4j-1.7.7.zip,然后解压缩,将slf4j-api-1.7.7,slf4j-nop-1.7.7.jar复制到kafka的libs目录中.

再次重启kafka生产者,现在没有报告错误.

来源:解决方案

Answer 2

您需要添加 SLF4j 日志记录实现。如果您使用 Maven 作为构建工具，请尝试将以下内容添加到您的 pom.xml 中，看看它是否有效。

    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>1.7.5</version>
    </dependency>