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

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

Producer

#####################################################################

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 = "myfirstTopic1";

    String key = "Key1";

    String value = "Value - 1";

       

       Properties props = new Properties();

       props.put("bootstrap.servers", "slc15atx.us.oracle.com:9092");

       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.");

    }

}

#########################################################################

Consumer

##########################################################################

package KafkaCode;

import java.util.*;

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

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

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

public class SimpleConsumer {

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

        String topicName = "myfirstTopic1";

        String groupName = "myfirstTopic1Group";

        Properties props = new Properties();

        props.put("bootstrap.servers", "slc15atx.us.oracle.com:9092,slc15atx.us.oracle.com:9093");

        props.put("group.id", groupName);

        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        @SuppressWarnings("resource")

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList(topicName));

        while (true){

                @SuppressWarnings("deprecation")

ConsumerRecords<String, String> records = consumer.poll(100);

                for (ConsumerRecord<String, String> record : records){

                        System.out.println("Value = " + String.valueOf(record.value()));

                }

        }

}

}

##########################################################################