5 KAFKA TRY: Practice Kafka [ Producer Java -> Multi Broker -> Consumer Console ]

5 KAFKA TRY: Practice Kafka [ Producer Java -> Multi Broker -> Consumer Console ]

We will try - 1 Node - Multiple brokers
#####################################-
conf/server.properties
Broker-id   (broker.id=0)
log-dir     log.dirs=/tmp/kafka-logs
ip:port     (listeners=PLAINTEXT://:9092)

conf/server1.properties
Broker-id  (broker.id=1)
log-dir     log.dirs=/tmp/kafka-logs2
ip:port    (listeners=PLAINTEXT://:9093)

conf/server2.properties
Broker-id  (broker.id=2)
log-dir     log.dirs=/tmp/kafka-logs3
ip:port    (listeners=PLAINTEXT://:9094)

Kafka : Running in Linux m/c (slc15atx.us.oracle.com)

Run Zookeper - bin/zookeeper-server-start.sh config/zookeeper.properties

Run Broker -
        bin/kafka-server-start.sh config/server.properties
        bin/kafka-server-start.sh config/server1.properties
        bin/kafka-server-start.sh config/server2.properties

Create topic - bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic NewMultiBrokerTopic --partitions 5 --replication-factor 3

bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic NewMultiBrokerTopic

Run consumer - 
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic NewMultiBrokerTopic

Producer : Write java in windows

------------------------------------------------------------------------------------------------------------


package KafkaCode;

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 SimpleProducer {
 public static void main(String[] args) throws Exception{
     
       String topicName = "NewMultiBrokerTopic";
    String key = "Key1";
    String value = "Value - MultiBroker - Test";
     
       Properties props = new Properties();
       props.put("bootstrap.servers", "slc15atx.us.oracle.com:9092,slc15atx.us.oracle.com:9093,slc15atx.us.oracle.com:9094");
       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 <>(props);

    ProducerRecord<String, String> record = new ProducerRecord<>(topicName,key,value);
    producer.send(record);     
       producer.close();
 
    System.out.println("SimpleProducer Completed.");
    }
}

------------------------------------------------------------------------------------------------------------