CODEWITHSUNDEEP
javaapache-kafkakafka-producer-api
Bob Wakefield
Bob Wakefield

Reputation: 4009

Kafka Producer does not appear to work properly

I have a Kafka producer that I've written in Java. It doesn't appear to work right even though it's basically a cut and past of example code. I would expect the output to be 10 messages to my cluster. Instead I get the Message Successful output but nothing actually goes to my cluster. I'm not certain where to start troubleshooting.

import java.util.Properties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class SimpleProducer {

 public static void main(String[] args) throws Exception{

    String topicName = "test_topic";
    Properties props = new Properties();
    props.put("bootstrap.servers", "skynet.local:6667");    
    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 = new KafkaProducer<String, String>(props);

    for(int i = 0; i < 10; i++)
       producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(i), Integer.toString(i)));
       System.out.println("Message sent successfully");
       producer.close();
 }
}

Upvotes: 1

Views: 2087

Answers (1)

南山竹
南山竹

Reputation: 524

Because some Environments uncleared, so I'll try to answered your question base on your Kafka Server is working on port 6667 already.

Your code may be need adjust at 2 palces (someone can help me improve it):

props.put("linger.ms", 1); // set to 0 let Producer can send message immediately

and here, drop producer.close(); out of for loop:

 Producer<String, String> producer = new KafkaProducer<String, String>(props, new StringSerializer(), new StringSerializer());
 for(int i = 0; i < 10; i++) {
   Future<RecordMetadata> f = producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(i), Integer.toString(i)));
   System.out.println(f.get()); // don't do that in your Production, here just for debugging purpose.
 }
 producer.close();

One more thing, you can run kafka-console-consumer.sh and kafka-console-producer.sh before your testing to confirm your Kafka server and your SimpleProducer is working already. Kafka 0.10.x configuration parameters at Kafka Producer Configuration Parameters

Upvotes: 2

Related Questions