Apache Kafka 프로듀서 소비자 API 문제

Question

카프카 프로듀서 및 소비자를 테스트하기위한 간단한 Java API를 만들려고합니다. Mac 컴퓨터의 별도 터미널에서 생산자와 소비자를 실행할 때 작동합니다. 하지만이 오류, 자바 API 코드를 사용하여 점점 카프카 서버에 연결을 시도 : 스레드에서

예외를 kafkatest2.ProducerTest.main에서 java.lang.NullPointerException이 "주"(ProducerTest.java:34)

생산자 코드 :

package kafkatest2; 
import java.util.Properties; 
import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.Producer; 
import org.apache.kafka.clients.producer.ProducerRecord; 
public class ProducerTest { 
    public static void main(String[] args) { 
    Properties props = new Properties(); 
    props.put("bootstrap.servers", "localhost:9092"); 
    props.put("acks", "all"); 
    props.put("retries", 0); 
    props.put("batch.size", 16384); 
    props.put("linger.ms", 1); 
    props.put("buffer.memory", 33554432); 
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer"); 
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer"); 

    Producer<String, String> producer = null; 
    try { 
      producer = new KafkaProducer<>(props); 
     for (int i = 0; i < 10; i++) { 
     String msg = "Message " + i; 
     producer.send(new ProducerRecord<String, String>("tested", msg)); 
     System.out.println("Sent:" + msg); 
     } 
    } catch (Exception e) { 
     e.printStackTrace(); 

    } finally { 
    System.out.println("last"); 
     producer.close(); 
    } 

    } 

} 

소비자 코드 :

package kafkatest2; 

import java.util.Arrays; 
import java.util.Properties; 

import org.apache.kafka.clients.consumer.ConsumerRecord; 
import org.apache.kafka.clients.consumer.ConsumerRecords; 
import org.apache.kafka.clients.consumer.KafkaConsumer; 

public class ConsumerTest { 

    public static void main(String[] args) { 
    Properties props = new Properties(); 
    props.put("zookeeper.connect", "127.0.0.1:2181"); 
    props.put("group.id", "test-consumer-group"); 
    props.put("enable.auto.commit", "true"); 
    props.put("auto.commit.interval.ms", "1000"); 
    props.put("auto.offset.reset", "earliest"); 
    props.put("session.timeout.ms", "30000"); 
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 

    KafkaConsumer<String, String> kafkaConsumer1 = new KafkaConsumer<>(props); 
    kafkaConsumer1.subscribe(Arrays.asList("tested")); 
    while (true) { 
     ConsumerRecords<String, String> records = kafkaConsumer1.poll(10); 
     for (ConsumerRecord<String, String> record : records) { 
     System.out.println("Partition: " + record.partition() + " Offset: " + record.offset() + " Value: " + record.value() + " ThreadID: " + Thread.currentThread().getId()); 
     } 
    } 

    } 

} 

,536,913,632 10

내가 무엇을 놓치고 있는지 알려주시겠습니까 ?? 구성 값에 문제가 있습니까 ??

덕분에, 또는 Vipul

Answer 1

스택 추적은 생산자 코드에 NPE를 나타냅니다 producer = new KafkaProducer<>(props);는 경우는 예외를 throw

Exception in thread "main" java.lang.NullPointerException at 
kafkatest2.ProducerTest.main(ProducerTest.java:34) 

이 발생할 수 있습니다. 그렇다면 finally 블록을 입력 할 때 로컬 변수 producer이 정의되지 않으므로 producer.close()을 호출하면 NPE가 반환됩니다. 전화를 if (producer != null) 블록으로 닫으십시오.