Kafka: сообщение на консоль отсутствует после сообщения, отправленного Java-производителем

Question

Я новичок в Kafka. Я создал java-продюсер на своей локальной машине и установил брокера Kafka на другой машине, скажем M2, в сети (я могу подключиться к SSH, подключиться к этой машине). На стороне Продюсера в консоли Eclipse я получаю сообщение «Сообщение отправлено». Но когда я проверяю потребитель консоли на машине M2, я не вижу эти сообщения.

Мой ява код производитель:

import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.ProducerConfig; 
import org.apache.kafka.clients.producer.ProducerRecord; 


import java.util.HashMap; 
import java.util.Map; 

public class KafkaMessageProducer { 

    /** 
    * @param args 
    */ 
    public static void main(String[] args) { 

     KafkaMessageProducer reportObj = new KafkaMessageProducer(); 
     reportObj.send(); 

    } 

    public void send(){ 

     Map<String, Object> config = new HashMap<String, Object>(); 
     config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "135.113.133.60:9092"); 
     config.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
     config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 

     KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config); 
     int maxMessages = 5; 
     int count = 0; 
     while(count < maxMessages){ 
      producer.send(new ProducerRecord<String, String>("test", "msg", "message --- #"+count++)); 
      System.out.println("Message send.."+count); 
     } 
     producer.close(); 
    } 

} 

Можете ли вы, пожалуйста, дайте мне знать, где я буду неправильно? Я могу отправлять сообщения локально на машине M2 от производителя консоли. Примечание. Даже когда я меняю IP-адрес на полное имя хоста Kafka Broker, он по-прежнему имеет такую ​​же проблему.

Обновление: Я также думаю, что Продюсер может подключиться к брокеру Kafka и отправлять сообщения, но Kafka Broker не передает эти сообщения потребителю. Если я изменю IP-адрес или порт на Zookeeper (который работает на том же узле, что и Kafka Broker), и просмотрите журнал Zookeeper, он получит команду «Продюсер» и затем отклонит сеанс.

Update2: Я создал банку производителя и запустил эту банку на машине M2, и она сработала. Таким образом, кажется, что что-то не так с тем, как Продюсер пытается подключиться к брокерам Kafka. Не уверен, в чем проблема.

Answer 1

Наконец-то я нашел ответ, и я отправляю его здесь, если у кого-то другая проблема. Если вы пытаетесь подключиться удаленно, используйте параметр брокера Kafka advertized.hostname. Это сработало для меня.

Answer 2

Вы можете использовать код следующим образом, чтобы прочитать информацию метаданных для темы kafka, чтобы узнать, получил ли брокер сообщения. Это может помочь отладки.

SimpleConsumer consumer = new SimpleConsumer(broker.host(), broker.port(), 100000, 
     64 * 1024, "your_group_id"); 
List<String> topics = new ArrayList<>(); 
topics.add(topic); 
TopicMetadataRequest req = new TopicMetadataRequest(topics); 

TopicMetadataResponse resp = simpleConsumer.send(req); 
if (resp.topicsMetadata().size() != 1) { 
    throw new RuntimeException("Expected one metadata for topic " 
     + topic + " found " + resp.topicsMetadata().size()); 
} 

TopicMetadata topicMetaData = resp.topicsMetadata().get(0); 

Answer 3

Подобно тому, как идея для отладки - попробовать producer.send(/* record */).get(); То есть, ждать результата от Future возвращенного из метода send(). Может быть, на стороне производителя есть исключение, и это просто игнорируется в фоновом режиме.