Kafka

A simple Kafka project structure

5 Easy steps !

1. Add the dependency in pom.xml if its a maven project.

        
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.3.0</version>
</dependency>
    

2. At first let’s declare an interface having Kafka configuration parameters as Constants.

        
package com.programmertoday.kafka;

public interface KafkaConstants {

  public static String KAFKA_BROKERS = "localhost:9092";
  public static Integer MESSAGE_COUNT = 10;
  public static String CLIENT_ID = "client_1";
  public static String TOPIC_NAME = "demotopic";
  public static String GROUP_ID_CONFIG = "consumerGroup_1";
  public static Integer MAX_NO_MESSAGE_FOUND_COUNT = 10;
  public static String OFFSET_RESET_LATEST = "latest";
  public static String OFFSET_RESET_EARLIER = "earliest";
  public static Integer MAX_POLL_RECORDS = 1;

}
    

3. Create a producer class – let’s call it “KafkaProducerClass”

        
package com.programmertoday.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

public class KafkaProducerClass {

	public static Producer<Long, String> createProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.KAFKA_BROKERS);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, KafkaConstants.CLIENT_ID);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class.getName()); // for custom partitions
        return new KafkaProducer<>(props);
    }
}
        

4. Create a consumer class – let’s call it “KafkaConsumerClass”

        
package com.programmertoday.kafka;

import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerClass {

    public static Consumer<Long, String> createConsumer() {
          Properties props = new Properties();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.KAFKA_BROKERS);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, KafkaConstants.GROUP_ID_CONFIG);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, KafkaConstants.MAX_POLL_RECORDS);
          props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, KafkaConstants.OFFSET_RESET_EARLIER);
          Consumer<Long, String> consumer = new KafkaConsumer<>(props);
          consumer.subscribe(Collections.singletonList(KafkaConstants.TOPIC_NAME));
          return consumer;
      } 
}
    

5. Create a Kafka test client – with main method in it and we poll the records.

        
package com.programmertoday.kafka;

import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class KafkaClient {

  public static void main(String[] args) {

    System.out.println("Kakfa Test in progress...........");
    executeProducer(); // if running producer, please commit the below line.
    //executeConsumer(); // if running consumer, please comment to above line.
  }

  static void executeConsumer() {
    Consumer<Long, String> consumer = KafkaConsumerClass.createConsumer();
    int noMessageFound = 0;
    while (true) { // condition set to true so that it keeps on running
      ConsumerRecords<Long, String> consumerRecords = consumer.poll(1000);
      // 1000 is the time in milliseconds consumer will wait if no record is found at
      // broker.
      if (consumerRecords.count() == 0) {
        noMessageFound++;
        if (noMessageFound > KafkaConstants.MAX_NO_MESSAGE_FOUND_COUNT)
          // If no message found count is reached to threshold exit loop.
          break;
        else
          continue;
      }
      // to print each record.
      consumerRecords.forEach(record -> {
        System.out.println("Record Key " + record.key());
        System.out.println("Record value " + record.value());
        System.out.println("Record partition " + record.partition());
        System.out.println("Record offset " + record.offset());
      });
      // commits the offset of record to broker.
      consumer.commitAsync();
    }
    consumer.close(); // to close the consumer
  }

  static void executeProducer() {
    Producer<Long, String> producer = KafkaProducerClass.createProducer();
    for (int index = 0; index < KafkaConstants.MESSAGE_COUNT; index++) {
      ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(KafkaConstants.TOPIC_NAME,
          "This is a kafka record " + index);
      try {
        RecordMetadata metadata = producer.send(record).get();
        System.out.println("Record sent with key " + index + " to partition " + metadata.partition()
            + " with offset " + metadata.offset());
      } catch (ExecutionException e) {
        System.out.println("Error in sending record");
        System.out.println(e);
      } catch (InterruptedException e) {
        System.out.println("Error in sending record");
        System.out.println(e);
      }
    }
  }

}
    

Summary

Now you know how to create kafka producer, kafka consumer in java, we hope you enjoyed learning this.

Follow These Very Simple Steps to install and start Kafka

Prerequisite: Download and install jre

Step 1 : Download Kafka from official website( https://kafka.apache.org/downloads )

Step 2 : Edit the two configuration files

• Server.properties
• Zookeeper.properties

Step 3 : Start Zookeeper and then start Kafka

To Test :

Step 4 : Create a topic, produce message on kafka topic, consume kafka topic

Let’s Begin !

1. Download kafka zip from the official website and then unzip it, you will see a folder structure like this.

2. Edit two config property files

Change from :
log.dirs=/tmp/kafka-logs
to
log.dirs=E:/kafka/kafka-logs

Change from :
dataDir=/tmp/zookeeper
To
dataDir=E:/kafka/zookeeper

3. Start zookeeper and then start kafkaStart zookeeper server

Goto > Kafka directory & run zookeeper

E:\softwares\kafka_2.12-2.3.0>.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

Start kafka server

E:\softwares\kafka_2.12-2.3.0>.\bin\windows\kafka-server-start.bat .\config\server.properties

You will see the message below if the start is successful : [2019-08-25 11:46:05,913] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

You will see the log directories and log files got created in the logs path given earlier in properties files.

4. Create Topic > produce message on topic > consume message from topic

Example Topic name : “demotopic”Create topic

E:\softwares\kafka_2.12-2.3.0>.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic demotopic

This is to create a topic

List Kafka topics

E:\softwares\kafka_2.12-2.3.0>.\bin\windows\kafka-topics.bat --list --zooke eper localhost:2181

List all kafka topicsProduce message on “demotopic”

E:\softwares\kafka_2.12-2.3.0>.\bin\windows\kafka-console-producer.bat --br oker-list localhost:9092 --topic demotopic

This produced message on a topic

Consume message from “demotopic”

E:\softwares\kafka_2.12-2.3.0>.\bin\windows\kafka-console-consumer.bat --bo otstrap-server localhost:9092 --topic demotopic --from-beginning

This is to consume a topic

Delete a topic – demotopic

E:\softwares\kafka_2.12-2.3.0>.\bin\windows\kafka-topics.bat --zookeeper lo calhost:2181 --delete --topic demotopic

Summary

Now you know how to install kafka zookeeper on your windows system and also you have learnt how to create topic , list topic, produce message and consume message from a topic and last but not least delete a topic.

Read next article to know how to create Producer Consumer code in JAVA.

Cluster Architecture of Apache Kafka

Apache Kafka Main Components

Cluster

It is a group of computers , each executing same instance of kafka broker.

Broker

It is just a meaningful name given to the kafka server, kafka producer does not directly interact with the consumer, they use kafka broker as the agent or broker to interact. In a cluster there can be more than one brokers.

Brokers are stateless, hence to maintain the cluster state they use ZooKeeper.

Zookeeper

ZooKeeper is used for managing and coordinating Kafka broker. ZooKeeper service is mainly used to notify producer and consumer about the presence of any new broker in the Kafka cluster system or failure of the broker in the Kafka cluster system. As per the notification received by the Zookeeper regarding presence or failure of the broker then producer and consumer takes decision and starts coordinating their task with some other broker.

Producers

Producer is a component which pushes data to the brokers, it doesn’t wait for acknowledgement from the brokers rather sends data as fast as the brokers can handle. There can be more than one producers depending on the use case.

Consumers

Since Kafka brokers are stateless, which means that the consumer has to maintain how many messages have been consumed by using partition offset. If the consumer acknowledges a particular message offset, it implies that the consumer has consumed all prior messages. The consumer issues an asynchronous pull request to the broker to have a buffer of bytes ready to consume. The consumers can kind of rewind or skip to any point in a partition simply by supplying an offset value. Consumer offset value is notified by ZooKeeper.

Kafka topic

A kafka topic is a logical channel to which producers publish messages and from which the consumers receive messages.

A topic name must be unique so that it is identifiable by both producer and consumer, there can be any number of topics, we cannot modify the data once published.

A Topic may contain any number of partitions as shown in the picture above.

Partitions in kafka

As you know broker store data of a topic, this data can be huge, break the data into two or more parts and distribute it to multiple computers.

In a Kafka cluster, Topics are split into Partitions and also replicated across brokers.

One can also add a key to the message to get ensured that all the messages with this key will end up in the same partition if the message is produced with the key. Because of this kafka also offers message sequence guarantee.

Otherwise without a key data is written to partitions randomly.

Offset

It is the sequence id given to a message in a partition, an offset is local to a partition, There can be any number of partitions, with no limitations to it.

— partition 1

— partition 2

Each partition sits on a single machine.

Note: How to directly locate a message ?

You need to know 3 things:

• Topic name
• Partition number
• Offset

Topic replication factor

It is always a good design decision to have a replication factor of a topic. It helps when a broker goes down the replica will still have the topic data. For example if the replication factor is 2 then a broker will have atleast one additional copy other than the primary.

Replication takes places at partition level only.

There has to be a leader among Brokers for a given partition and that will be only one. The number of replication factor cannot be greater than the number of available brokers.

Consumer Group

Scenario : when hundreds of producers produce data to a single consumer, it’s hard to manage its volume and velocity.

Partitioning and consumer group is a tool for scalability, Maximum number of consumers in a group is equal to the total number of partitions you have on the topic.

Kafka doesn’t allow more than 2 consumers to read data from the same partition.

Also one consumer group will have one unique group id.

Summary

We learn about Kakfa features, its uses, usecases and core apis, Hope you liked it !