Kafka Producer and Consumer Configuration

Kafka provides a flexible and powerful configuration mechanism for producers and consumers to optimize performance and ensure reliable message processing.

Producer Configuration

When configuring a Kafka producer, you can set various properties to fine-tune its behavior:

• bootstrap.servers: Specifies the list of brokers in the Kafka cluster that the producer should connect to.

• key.serializer: Specifies the serializer class for key objects.

• value.serializer: Specifies the serializer class for value objects.

• acks: Specifies the number of acknowledgments the producer requires from the broker before considering a message as sent.

• retries: Specifies the number of times the producer should retry sending a message if an error occurs.

Consumer Configuration

Similarly, configuring a Kafka consumer involves setting properties to control its behavior:

• bootstrap.servers: Specifies the list of brokers in the Kafka cluster that the consumer should connect to.

• key.deserializer: Specifies the deserializer class for key objects received from Kafka.

• value.deserializer: Specifies the deserializer class for value objects received from Kafka.

• group.id: Specifies the consumer group to which the consumer belongs.

• auto.offset.reset: Specifies the strategy to use when there is no initial offset or if the current offset does not exist on the server.

Here's an example of a simple Java program that configures a Kafka producer:

1import org.apache.kafka.clients.producer.Producer;
2import org.apache.kafka.clients.producer.ProducerRecord;
3import org.apache.kafka.clients.producer.KafkaProducer;
4
5import java.util.Properties;
6
7public class KafkaProducerExample {
8  public static void main(String[] args) {
9    // Create a new producer
10    KafkaProducer<String, String> producer = new KafkaProducer<>(getProducerProperties());
11
12    // Send a message
13    String message = "Hello, Kafka!";
14    ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", message);
15    producer.send(record);
16
17    // Flush and close the producer
18    producer.flush();
19    producer.close();
20  }
21
22  private static Properties getProducerProperties() {
23    Properties props = new Properties();
24    props.put("bootstrap.servers", "localhost:9092");
25    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
26    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
27    return props;
28  }
29}

xxxxxxxxxx

class Main {

    public static void main(String[] args) {

    // replace with your Java logic here

    for(int i = 1; i <= 100; i++) {

      if(i % 3 == 0 && i % 5 == 0) {

          System.out.println("FizzBuzz");

      } else if(i % 3 == 0) {

          System.out.println("Fizz");

      } else if(i % 5 == 0) {

          System.out.println("Buzz");

      } else {

          System.out.println(i);

      }

    }    

  }

}