Kafka ConsumerDemo/producer.java

	import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord;
	import org.apache.kafka.clients.consumer.ConsumerRecords;
	import org.apache.kafka.clients.consumer.KafkaConsumer;
	import org.apache.kafka.common.serialization.StringDeserializer;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import java.time.Duration;
	import java.util.Arrays;
	import java.util.Properties;
	public class ConsumerDemo {
	public static void main(String[] args) {
	Logger logger = LoggerFactory.getLogger(ConsumerDemo.class.getName());
	String bootstrapServers = "127.0.0.1:9092";
	String groupId = "my-fourth-application";
	String topic = "first_topic";
	// create consumer configs
	Properties properties = new Properties();
	properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
	properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
	properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
	properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
	properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
	// create consumer
	KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
	// subscribe consumer to our topic(s)
	consumer.subscribe(Arrays.asList(topic));
	// poll for new data
	while(true){
	ConsumerRecords<String, String> records =
	consumer.poll(Duration.ofMillis(100)); // new in Kafka 2.0.0
	for (ConsumerRecord<String, String> record : records){
	logger.info("Key: " + record.key() + ", Value: " + record.value());
	logger.info("Partition: " + record.partition() + ", Offset:" + record.offset());
	}
	}
	}
	} import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public class ProducerDemo { public static void main(String[] args) { String bootstrapServers = "127.0.0.1:9092"; // create Producer properties Properties properties = new Properties(); properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // create the producer KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties); // create a producer record ProducerRecord<String, String> record = new ProducerRecord<String, String>("first_topic", "hello world"); // send data - asynchronous producer.send(record); // flush data producer.flush(); // flush and close producer producer.close(); } }