CodeSample

Producer code:

package com.org;
import java.util.Properties;
import org.apache.log4j.BasicConfigurator;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class ProducerDemo {
public static void main(String[] args) throws Exception{
BasicConfigurator.configure();

Properties props = new Properties();
props.put("bootstrap.servers", "hostname.org.com:9092");
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");

@SuppressWarnings("unchecked")
Producer<String, String> producer = (Producer<String, String>) new KafkaProducer(props);
producer.send(new ProducerRecord<String, String>("test", "hello"));

producer.close();
}
}



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

consumer code java :

package com.org;

import java.util.Arrays;
import java.util.Properties;

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

import org.apache.log4j.BasicConfigurator;

public class ConsumerDemo {

public static void main(String[] args) throws Exception{
BasicConfigurator.configure();

Properties props = new Properties();
props.put("bootstrap.servers", "hostname.org.com:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
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<String, String>(props);

consumer.subscribe(Arrays.asList("test"));

//print the topic name
System.out.println("Subscribed to topic " + "test");
int i = 0;

while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)

// print the offset,key and value for the consumer records.
System.out.printf("offset = %d, key = %s, value = %s\n",
record.offset(), record.key(), record.value());
}
}
}



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



Python script for producer and consumer:

print("I am in kafka demo")

from kafka import KafkaProducer
from kafka import KafkaConsumer

producer = KafkaProducer(bootstrap_servers='hostname.org.com:9092')
producer.send('mkafkatopic', b'msg %d' % 206)
producer.send('mkafkatopic',b'hello kafka')

metrics = producer.metrics()
print(metrics)

consumer = KafkaConsumer("mkafkatopic", bootstrap_servers="hostname.org.com:9092",
auto_offset_reset='earliest')

for message in consumer:
print(message.value)