Брокер Kafka может быть недоступен, исключение ⇐ JAVA

[Anonymous]

Я столкнулся со странной проблемой с моим продюсером Kafka. Я использую серверную/клиентскую версию Kafka-0.11.
У меня есть один Zookeper и один брокерский узел Kafka. Кроме того, я создал тему «События» с тремя разделами:

Topic:events PartitionCount:3 ReplicationFactor:1 Configs:
Topic: events Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: events Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: events Partition: 2 Leader: 0 Replicas: 0 Isr: 0


В моем Java-коде я создаю производителя со следующими свойствами:

Properties props = new Properties();
props.put(BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
props.put(MAX_BLOCK_MS_CONFIG, 30000);
props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(PARTITIONER_CLASS_CONFIG, KafkaCustomPartitioner.class);
this.producer = new KafkaProducer(props);


Кроме того, я добавил обратный вызов к методу Producer#send(), который добавляет сообщение о сбое в очередь, которая повторяется другим потоком «повторной отправки» в цикле:

this.producer.send(producerRecord, new ProducerCallback(producerRecord.value(), topic));

private class ProducerCallback implements Callback {
private final String message;
private final String topic;

public ProducerCallback(String message, String topic) {
this.message = message;
this.topic = topic;
}

@Override
public void onCompletion(RecordMetadata metadata, Exception ex) {
if (ex != null) {
logger.error("Kafka producer error. Topic: " + topic +
".Message will be added into failed messages queue.", ex);
failedMessagesQueue.enqueue(SerializationUtils.serialize(new FailedMessage(topic, message)));
}
}
}

private class ResenderThread extends Thread {
private volatile boolean running = true;

public void stopGracefully() {
running = false;
}

@Override
public void run() {
while (running) {
try {
byte[] val = failedMessagesQueue.peek();
if (val != null) {
FailedMessage failedMessage = SerializationUtils.deserialize(val);
ProducerRecord record;
if (topic.equals(failedMessage.getTopic())) {
String messageKey = generateMessageKey(failedMessage.getMessage());
record = createProducerRecordWithKey(failedMessage.getMessage(), messageKey, failedMessage.getTopic());
} else {
record = new ProducerRecord(failedMessage.getTopic(), failedMessage.getMessage());
}
try {
this.producer.send(record).get();
failedMessagesQueue.dequeue();
} catch (Exception e) {
logger.debug("Kafka message resending attempt was failed. Topic " + failedMessage.getTopic() +
" Partition. " + record.partition() + ". " + e.getMessage());
}
}

Thread.sleep(200);
} catch (Exception e) {
logger.error("Error resending an event", e);
break;
}
}
}
}


Все работает нормально, пока я не решил протестировать сценарий уничтожения/перезапуска брокера Kafka:

Я закрыл свой узел брокера Kafka и отправил 5 сообщений с помощью моего производителя Kafka. Следующее сообщение было зарегистрировано моим приложением-производителем:

....the application works fine....
// kafka broker was killed
2017-11-10 09:20:44,594 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:44,646 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:44,700 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:44,759 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:44,802 WARN [org.apache.kafka.clients.NetworkClient] -
// sent 5 message using producer. message were put to the failedMessagesQueue and "re-sender" thread started resending
2017-11-10 09:20:44,905 ERROR [com.inq.kafka.KafkaETLService] -
....
2017-11-10 09:20:45,070 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:45,129 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:45,170 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:45,217 WARN [org.apache.kafka.clients.NetworkClient] -

// kafka broker was restarted, some strange errors were logged
2017-11-10 09:20:51,103 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:51,205 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:51,308 WARN [org.apache.kafka.clients.NetworkClient] -
2017-11-10 09:20:51,114 WARN [org.apache.kafka.clients.producer.internals.Sender] -
2017-11-10 09:20:51,114 ERROR [com.inq.kafka.KafkaETLService] -
2017-11-10 09:20:52,485 WARN [org.apache.kafka.clients.NetworkClient] -
// messages were succesfully re-sent and received by consumer..


Как мне избавиться от этих журналов (которые регистрируются каждые 100 мс, когда брокер Kafka не работает):

[org.apache.kafka.clients.NetworkClient] -


Почему я получаю следующие ошибки после запуска брокера Kafka (я не менял никакие реквизиты сервера и не менял тему). Мне кажется, что эти ошибки являются результатом некоторого процесса синхронизации между Zookeeper и Kafka во время запуска брокера, потому что через некоторое время производитель успешно отправляет мои сообщения. Я ошибаюсь?:

[org.apache.kafka.clients.NetworkClient] -
Received unknown topic or partition error in produce request on partition events-0. The topic/partition may not exist or the user may not have Describe access to it.


Подробнее здесь: https://stackoverflow.com/questions/472 ... -exception