Блог

У меня есть приведенный ниже код kafka producer, и когда я его запускаю, я не вижу никаких ошибок, и записи не отображаются в консоли пользователя. Я использую https://kafka.apache.org/quickstart чтобы запустить zookeeper, брокер. Я создал тему и запустил потребителя.

 import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class SampleProducerCreator {

    Properties properties =  new Properties();

    private void init(){
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("kafka.topic.name", "quickstart-events");
        KafkaProducer<String, String> producer = new KafkaProducer<>(this.properties,
                new StringSerializer(), new StringSerializer());
        for(int i=0; i<4 ; i  ){
            String payload = "Test";
            ProducerRecord<String, String> record = new ProducerRecord<>(properties.getProperty("kafka.topic.name"), payload);
            producer.send(record);
        }
        producer.close();
    }

   public static void main(String[] args){
        SampleProducerCreator sampleProducerCreator = new SampleProducerCreator();
        sampleProducerCreator.init();
   }
}
  

Могут возникнуть проблемы с подключением к брокеру. Если в ваш java-проект не добавлена реализация slf4j, журналы не будут распечатаны.

Кроме того, producer.send(record) возвращает будущее. Вы можете использовать это будущее для блокировки и ожидания ответа, или лучше использовать альтернативный send(record, callback) javadoc для распечатки исключения или записи метаданных, возвращенных брокером.