Kafka

https://kafka.apache.org/

A distributed streaming platform.

1. can be used as an enterprise messaging system

2. can be used it as a stream processing platform. Kafka gives a stream, and we can plug in a processing framework.

3. Also provides connectors to export and import bulk data from databases and other systems.

https://kafka.apache.org/images/kafka_diagram.png

Install/Unzip Apache Kafka

tar -zxvf kafka_2.11-0.10.1.0.tgz

Start Kafka Server

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

bin/kafka-server-start.sh config/server.properties

kafka-topics.sh is a tool to manage a Kafka

Create Kafka Topic (replication factor 3, and 2 partitions)

bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic TopicName --partitions 2 --replication-factor 2

describe Kafka Topic

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

• ISR is a list of In Sync Replicas

Start Kafka Producer and Consumer

console producer

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TopicName

console consumer

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

Broker Configurations

// 3 essential configurations are the following:

broker.id
port
log.dirs

zookeeper.connect
delete.topic.enable
auto.create.topics.enable
default.replication.factor
num.partitions
log.retention.ms
log.retention.bytes

complete list : https://kafka.apache.org/documentation/#brokerconfigs

Build.sbt

    name := "KafkaTest"

    libraryDependencies ++= Seq(
        "org.apache.kafka" % "kafka-clients" % "0.10.1.0"
        exclude("javax.jms", "jms")
        exclude("com.sun.jdmk", "jmxtools")
        exclude("com.sun.jmx", "jmxri")
        exclude("org.slf4j", "slf4j-simple")
    )

Simple Kafka producer

import java.util.*;
    import org.apache.kafka.clients.producer.*;                                    
    public class TestProducer {                                    
            public static void main(String[] args) throws Exception{

                String key = "Key1";                
                String value = "Value1";
                String topicName = "KafkaTestTopic";

                Properties props = new Properties();                
                props.put("bootstrap.servers", "localhost:9092,localhost:9093");
                props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");         
                props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

                Producer producer = new KafkaProducer<>(props);
                ProducerRecord record = new ProducerRecord<>(topicName,key,value);

                producer.send(record);                
                producer.close();
        }
    }