Using Kafka in 2022
Tags: java, productivity, programming, architecture
Published: 2022-03-10 03:09:52

*In case it helped :)*
<a href="https://www.buymeacoffee.com/aseemwangoo" target="_blank"><img src="https://www.buymeacoffee.com/assets/img/custom_images/orange_img.png" alt="Pass Me A Coffee!!" style="height: 41px !important;width: 174px !important;box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;-webkit-box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;" ></a>

<!-- wp:paragraph -->
<p>We will cover briefly:</p>
<!-- /wp:paragraph -->

<!-- wp:list {"ordered":true} -->
<ol><li>Intro to Kafka</li><li>Setup Kafka on machine</li><li>Create Kafka Producer</li><li>Create Kafka Consumer</li><li>Running Producer and Consumer</li></ol>
<!-- /wp:list -->

{% youtube 0d94HwVjfLQ %}

<!-- wp:heading {"level":4} -->
<h4>Intro to&nbsp;Kafka</h4>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>As per the <a href="https://kafka.apache.org/intro" rel="noreferrer noopener" target="_blank">docs</a>, Kafka is an event streaming platform that can be used for collecting, processing, storing, and analyzing data at scale. <strong>Kafka </strong>is used by <a href="https://kafka.apache.org/powered-by" rel="noreferrer noopener" target="_blank">thousands of companies</a> including over 60% of the Fortune 100.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Kafka is known for its excellent performance, low latency, fault tolerance, and high throughput, it’s capable of handling thousands of messages per second.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"align":"center"} -->
<div class="wp-block-image"><figure class="aligncenter"><img src="https://cdn-images-1.medium.com/max/1600/0*QFtT6kh-s5SfhIWf.png" alt="Kafka Overview"/><figcaption>Kafka Overview</figcaption></figure></div>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>Usecases:</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Process payments and financial transactions in real-time,</li><li>Continuously capture and analyze sensor data from IoT devices,</li><li>Collect and immediately react to customer interactions and orders,</li><li>Track and monitor cars, trucks, fleets, and shipments in real-time</li></ul>
<!-- /wp:list -->

<!-- wp:heading {"level":4} -->
<h4><strong>Setup Kafka on&nbsp;machine</strong></h4>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>This section will help to set up Kafka on your machine.&nbsp;</p>
<!-- /wp:paragraph -->

<!-- wp:quote -->
<blockquote class="wp-block-quote"><p><strong>Note: We are using a mac.</strong></p></blockquote>
<!-- /wp:quote -->

<!-- wp:list -->
<ul><li>To install <code>Kafka</code> follow <a href="https://kafka.apache.org/downloads" rel="noreferrer noopener" target="_blank">here</a>. Navigate to the <strong>latest binary downloads</strong>, as of today, it’s <strong>Scala 2.13, </strong>and click on the link. It will open another webpage and click the link as below:</li></ul>
<!-- /wp:list -->

<!-- wp:image {"align":"center"} -->
<div class="wp-block-image"><figure class="aligncenter"><img src="https://cdn-images-1.medium.com/max/1600/1*CND4cBRTMob43AdMYHLyOQ.png" alt="Download link for Kafka"/><figcaption>Download link for Kafka</figcaption></figure></div>
<!-- /wp:image -->

<!-- wp:list -->
<ul><li>Extract the folder and navigate to the folder (<em>Downloads/kafka_2.12–3.1.0</em>) in our case.</li></ul>
<!-- /wp:list -->

<!-- wp:quote -->
<blockquote class="wp-block-quote"><p>Note: Follow the below terminal commands at the path where you extract the Kafka zip.&nbsp;</p></blockquote>
<!-- /wp:quote -->

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

<!-- wp:paragraph -->
<p><strong>Start Zookeeper</strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>The above command starts <code>Zookeeper</code>&nbsp;. It is primarily used to track the status of nodes in the Kafka cluster and maintain a list of Kafka topics and messages.</p>
<!-- /wp:paragraph -->

<!-- wp:quote -->
<blockquote class="wp-block-quote"><p>Note: Starting with v2.8, Kafka can be run without <code>Zookeeper</code>. However, this update isn’t ready for use in production</p></blockquote>
<!-- /wp:quote -->

<!-- wp:quote -->
<blockquote class="wp-block-quote"><p>Keep the above terminal open</p></blockquote>
<!-- /wp:quote -->

<!-- wp:paragraph -->
<p><strong>Start Kafka broker service</strong></p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Open another terminal at the above path(<em>Downloads/kafka_2.12–3.1.0</em>) and type</li></ul>
<!-- /wp:list -->

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

<!-- wp:paragraph -->
<p>This starts the <code><em>Kafka Broker</em></code>. A <em>Kafka cluster</em> is made up of multiple Kafka Brokers. Each Kafka Broker has a unique ID (number). Kafka Brokers contain topic log partitions. Connecting to one broker bootstraps a client to the entire Kafka cluster. A Kafka cluster can have, 10, 100, or 1,000 brokers in a cluster if needed.</p>
<!-- /wp:paragraph -->

<!-- wp:quote -->
<blockquote class="wp-block-quote"><p>Keep this terminal open</p></blockquote>
<!-- /wp:quote -->

<!-- wp:paragraph -->
<p><strong>Create a topic</strong></p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Open a new terminal at the above path and type</li></ul>
<!-- /wp:list -->

```bash
sh bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic
```

<!-- wp:paragraph -->
<p>Kafka produces events across different machines, these events are organized and stored inside <code>topics</code>&nbsp;</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>In the above command, we are saying our&nbsp;</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>server is <code>localhost:9092</code> and</li><li>topic created is <code>test-topic</code>&nbsp;</li></ul>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p><strong>Verify the topic created&nbsp;</strong></p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Open a new terminal at the above path and type</li></ul>
<!-- /wp:list -->

```bash
sh bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
```

<!-- wp:paragraph -->
<p>This should show you the <code>test-topic</code></p>
<!-- /wp:paragraph -->

<!-- wp:heading {"level":4} -->
<h4>Create Kafka&nbsp;Producer</h4>
<!-- /wp:heading -->

<!-- wp:quote -->
<blockquote class="wp-block-quote"><p>Note: Kafka Broker should be running in the terminal</p></blockquote>
<!-- /wp:quote -->

<!-- wp:paragraph -->
<p><strong>Install dependencies</strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>We will be using <a href="https://www.eclipse.org/downloads/" target="_blank" rel="noreferrer noopener">Eclipse IDE</a> for creating our Kafka Producer</p>
<!-- /wp:paragraph -->

<!-- wp:image {"align":"center"} -->
<div class="wp-block-image"><figure class="aligncenter"><img src="https://cdn-images-1.medium.com/max/1600/0*6onkDUNIQdfeQlst" alt="Kafka Producer"/><figcaption>Kafka Producer</figcaption></figure></div>
<!-- /wp:image -->

<!-- wp:list -->
<ul><li>Create a Maven Project using the eclipse.</li><li>Go to the pom.xml, put in the following, and update your maven project</li></ul>
<!-- /wp:list -->

```xml
<dependencies> 
   
   <!-- Kafka Client -->
   <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.1.0</version>
   </dependency>
   
   <!-- Simple logging -->
   <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>2.0.0-alpha6</version>
   </dependency>
   <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-simple</artifactId>
    <version>2.0.0-alpha6</version>     
   </dependency>
</dependencies>
```
<!-- wp:paragraph -->
<p>This will add <code>Kafka Client</code> and <code>slf4j</code> to our project.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><strong>Create SampleProducer.java</strong></p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Create a class called <code>SampleProducer.java</code></li></ul>
<!-- /wp:list -->

```java
public class SampleProducer {
 
    
 public SampleProducer() throws InterruptedException {
  Properties properties = new Properties();
  properties.put("bootstrap.servers", "localhost:9092");
  properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
 properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
 
  properties.put("kafka.topic", "test-topic");
  
  KafkaProducer kafkaProducer = new KafkaProducer(properties);
  int i=0;
  
  try {
   while (true) {
    i++;
    ProducerRecord producerRecord = new ProducerRecord("test-topic","key","message-" +i);
    
    Thread.sleep(3000);
    kafkaProducer.send(producerRecord);
   }
  } finally {
   kafkaProducer.close();
  }    
    }
                  
}
```

<!-- wp:paragraph -->
<p>We initialize the <code>Properties</code> with the following keys and values&nbsp;</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>bootstrap.servers</code>&nbsp;: your local server host and port</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>key.serializer</code>&nbsp;: for sending the serialized key over the network.&nbsp;</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>value.serializer</code>&nbsp;: for sending the serialized value over the network.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>kafka.topic</code>&nbsp;: your Kafka topic</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Next, we create an instance of <code>KafkaProducer</code> with the properties (as specified above)</li><li>We need to create a <code>ProducerRecord</code>for sending the data to the Kafka producer.</li></ul>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p>This ProducerRecord takes in the <strong>topic name</strong>, <strong>key</strong>, and the <strong>value</strong> to be sent.</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>We send the record using <strong>kafkaProducer.&nbsp;</strong></li><li>This block of code is placed inside the while loop, and after every 3 seconds, it sends a record to the Kafka broker.</li></ul>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p><strong>Create KafkaProducerRunner.java</strong></p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>This will be class that invokes the above <code>SampleProducer</code></li></ul>
<!-- /wp:list -->

```java
public class KafkaProducerRunner {
 public static void main(String[] args) {
  SampleProducer sampleProducer = new SampleProducer();
 }
}
```
<!-- wp:heading {"level":4} -->
<h4>Create Kafka&nbsp;Consumer</h4>
<!-- /wp:heading -->

<!-- wp:quote -->
<blockquote class="wp-block-quote"><p>Note: Kafka Broker should be running in the terminal</p></blockquote>
<!-- /wp:quote -->

<!-- wp:list -->
<ul><li>Create a class called <code>SampleConsumer.java</code></li></ul>
<!-- /wp:list -->

```java
public class SampleConsumer {
 
 public SampleConsumer() {
  Properties properties = new Properties();
  properties.put("bootstrap.servers", "localhost:9092"); 
  properties.put("kafka.topic", "test-topic"); 
  properties.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
 properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  
 properties.put("group.id", "my-group");
  
  KafkaConsumer consumer = new KafkaConsumer(properties);
  consumer.subscribe(Arrays.asList(properties.getProperty("kafka.topic")));
        
       while(true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
          for (ConsumerRecord record : records)  {
           System.out.printf("partition = %s, offset = %d, key = %s, value = %s\n", 
             record.partition(), record.offset(), record.key(), record.value());
          }
       }              
  
 }  
}
```

<!-- wp:paragraph -->
<p>We initialize the <code>Properties</code> with the following keys and values</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>bootstrap.servers</code>&nbsp;: your local server host and port</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>kafka.topic</code>&nbsp;: your Kafka topic</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>key.deserializer</code>&nbsp;: for deserializing the key over the network.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>value.deserializer</code>&nbsp;: for deserializing the value over the network.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><code>group.id</code>&nbsp;: specify the group id, used only on the Consumer side.</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Next, we create an instance of <code>KafkaConsumer</code> with the properties (as specified above). This KafkaConsumer consumes the records from a Kafka cluster.</li><li>Next, we subscribe to the topic (which was created in the KafkaProducer) using <code>subscribe</code>&nbsp;</li><li>We get the data in the form of <code>ConsumerRecords</code> and we call <code>poll</code> every 100 ms in an infinite loop.</li><li>For each consumer record, we extract the <strong>partition, offset, key </strong>and <strong>value</strong></li></ul>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p><strong>Create KafkaConsumerRunner.java</strong></p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>This will be class that invokes the above <code>SampleConsumer</code></li></ul>
<!-- /wp:list -->

```java
public class KafkaConsumerRunner {
 public static void main(String[] args) {
  SampleConsumer sampleConsumer = new SampleConsumer();
 }
}
```

<!-- wp:heading {"level":4} -->
<h4>Running Producer and&nbsp;Consumer</h4>
<!-- /wp:heading -->

<!-- wp:paragraph -->
<p>If you followed the above steps correctly, you should have the below files</p>
<!-- /wp:paragraph -->

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/pk7fvgkzzd3bj1x5qcp0.png)

<!-- wp:paragraph -->
<p>Let’s run the <strong>KafkaProducerRunner first, </strong>we should see something like this</p>
<!-- /wp:paragraph -->

<!-- wp:image {"align":"center"} -->
<div class="wp-block-image"><figure class="aligncenter"><img src="https://cdn-images-1.medium.com/max/1600/1*9OKX-3DihIyWCO2-sUXVZA.png" alt="Kafka Producer logs"/><figcaption>Kafka Producer logs</figcaption></figure></div>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>Our Producer produces data every 3 seconds.</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li>Let’s run the <strong>KafkaConsumerRunner, </strong>we should see something like this</li></ul>
<!-- /wp:list -->

<!-- wp:image {"align":"center","width":840,"height":60} -->
<div class="wp-block-image"><figure class="aligncenter is-resized"><img src="https://cdn-images-1.medium.com/max/1600/1*CTrerBBLu3M-W_Dqhskmfw.png" alt="Kafka Consumer logs" width="840" height="60"/><figcaption>Kafka Consumer logs</figcaption></figure></div>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>Our consumer receives the data, every 3 seconds and we print it onto the console</p>
<!-- /wp:paragraph --> 

<!-- wp:paragraph -->
<p><a rel="noreferrer noopener" href="https://github.com/AseemWangoo/kafka-example" target="_blank"><em>Source code.</em></a>

{% youtube 0d94HwVjfLQ %}

*In case it helped :)*
<a href="https://www.buymeacoffee.com/aseemwangoo" target="_blank"><img src="https://www.buymeacoffee.com/assets/img/custom_images/orange_img.png" alt="Pass Me A Coffee!!" style="height: 41px !important;width: 174px !important;box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;-webkit-box-shadow: 0px 3px 2px 0px rgba(190, 190, 190, 0.5) !important;" ></a>