Q:

Apache Kafka Producer Consumer API Issue

I am trying to create a simple Java API to test Kafka Producer and Consumer. When I am running the producer and consumer on separate terminals on my Mac machine its working fine. But when I tried to connect to Kafka server using java api code, getting this error :
Exception in thread "main" java.lang.NullPointerException
    at kafkatest2.ProducerTest.main(ProducerTest.java:34)
Producer Code :
    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();
    }

  }

}

Consumer Code :
 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());
      }
    }

  }

}

Please let me know what am I missing?? is there some issue with the configuration values??
Thanks,
Vipul

A:

The stack trace indicates an NPE in your producer code:
Exception in thread "main" java.lang.NullPointerException at 
kafkatest2.ProducerTest.main(ProducerTest.java:34)

This can happen if producer = new KafkaProducer<>(props); throws an Exception. If so, when you enter the finally block, the local variable producer won't be defined so calling producer.close() will throw an NPE. Simply wrap the call to close in an if (producer != null) block.

