Ошибка Apache Kafka: не удалось отправить сообщение после 3 попыток

Я запускаю этот пример производителя 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 хост-сервер брокера от хоста-производителя.

  • Предоставлено advertised.host.name в server.properties(они называются сервером [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

Любая идея о том, как исправить эту ошибку?

Ответы

Ответ 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 server и создаю тестовую тему. Затем мне нужно запустить продюсера для тестовой темы. но есть ошибка:

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 в файле zip файла kafka включен только jar файл slf4j-api, они пропустили файл jar: slf4j-nop.jar, поэтому нам нужно перейти к http://www.slf4j.org, загрузите slf4j-1.7.7.zip, а затем разархивируйте его, скопируйте slf4j-api-1.7.7, slf4j-nop-1.7.7.jar в каталог kafkas libs.

Перезапустите продюсер kafka, теперь сообщение об ошибке не сообщается.

Источник: РЕШЕНИЕ

Ответ 2

Вам нужно добавить реализацию регистрации SLF4j. если вы используете maven в качестве инструмента сборки, попробуйте добавить это следующее в свой pom.xml и посмотреть, не работает ли он.

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

Ответ 3

Это решение исключения в исходном вопросе, заданном Кришем: "kafka.common.FailedToSendMessageException: не удалось отправить сообщения после 3 попыток".

Часто задаваемые вопросы здесь и здесь говорит, что ваше имя хоста должен быть установлен правильно. Я не испытал этого состояния. Но я нашел другое условие, когда производитель kafka выдает это сообщение об ошибке: когда ваш ключ раздела в изготовителе ошибочен. То есть, если у вас есть тема с одним разделом, то ключ раздела в производителю может быть либо нулевым (сообщение отправляется в произвольный раздел), либо 0 (разделы в кафке нумеруются начиная с 0). Если вы попытаетесь использовать ключ раздела 1, это исключение будет передано производителю. Или если у вас есть 3 раздела в теме, и вы используете ключ раздела 3 (ключ из 3 недействителен, поскольку допустимые номера разделов равны 0,1,2), это исключение выдается. Эта ошибка согласована, если номер раздела в методе send() производителя не соответствует диапазону разделов в теме. Я использовал kafka версии 0.8.2. API-интерфейс клиента, который я использовал, был пакетом kafka.javaapi.producer.Producer.

Ответ 4

Это может произойти, если клиент не может связаться с BOTH именем хоста и IP-адресом брокера kafka.

Сделайте запись для клиентов \etc\hosts или C:\Windows\System32\drivers\etc\hosts и решила эту проблему для меня.

Ответ 5

Я получил ошибку от apache kafka:

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

Моя настройка:

OS: Ubuntu 14.04
sbt: sbt launcher version 0.13.5
scala: Scala code runner version 2.9.2

Удалось исправить это с помощью следующих команд:

cd /home/el/apachekafka/kafka_2.10-0.8.1.1/libs
wget http://www.slf4j.org/dist/slf4j-1.7.7.tar.gz
tar -xvf slf4j-1.7.7.tar.gz
cd /home/el/apachekafka/kafka_2.10-0.8.1.1/libs/slf4j-1.7.7
cp slf4j-api-1.7.7.jar ..
cp slf4j-nop-1.7.7.jar ..

Затем запустите команду, и производитель не выдаст никакой ошибки.

Ответ 6

ForhDP kafka использует порт брокера: 6667

Для автономного kafka используйте порт брокера: 9092

Ошибка произошла из-за того порта, который мы использовали (HDP использует 6667, но мы использовали 9092)

bin/kafka-console-producer.sh -broker-list broker-ip: 9092 --topic test//не работает

bin/kafka-console-producer.sh -broker-list broker-ip: 6667 --topic test//working

ссылка: Производитель консоли Kafka Ошибка в песочнице Hortonworks HDP 2.3

Ответ 7

Не уверен, но одна возможность может заключаться в том, что тема не создана на Кафке.

Проверьте веб-интерфейс для kafka и убедитесь, что тема, которую вы используете, то есть "page_visits" для отправки данных создается там.

Если это не очень просто создать тему с помощью графического интерфейса.

Ответ 8

Я столкнулся с этой ошибкой в ​​Hortonworks HDP 2.2, где порт по умолчанию установлен на 6667. Если ваш сервер kafka работает на песочнице HDP, разрешение должно установить metadata.broker.list as 10.0.2.15:6667 Пожалуйста, следуйте этому коду.

            Properties props = new Properties();
            props.put("metadata.broker.list", "10.0.2.15:6667");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            //props.put("producer.type","async");
            props.put("request.required.acks", "1");
            ProducerConfig config = new ProducerConfig(props);

            Producer<String, String> producer = new Producer<String, String>(config);
            try{        
            producer.send(new KeyedMessage<String, String>("zerg.hydra", jsonPayload));

            producer.close();
        }catch(Exception e){
            e.printStackTrace();
        }