DEV Community: RaghuNandan Neerukonda

KAFKA - PRODUCER API WITH THIRD PARTY TOOLS

RaghuNandan Neerukonda — Tue, 23 Jun 2020 11:35:52 +0000

This is a continuation write-up to the previous post - Kafka - Playing with Consumer API, where I covered some peculiar use-cases of Consumer API.
Here, I would like to play with Producer API to create a message using third-party tools like MessagePack for serialization and KafDrop for monitoring

MessagePack

MessagePack is one of the best available schemaless serialization libraries to interchange data between heterogeneous applications. It is a binary for simple data structures and designed for efficient transmission over the wire.

Code Snippets

Following examples shows how to pack and unpack the common data types, to work with custom data types like datetime.datetime, check DateTime example

Simple Example

from msgpack import packb, unpackb

data = {'a': 5}
msg_packet = packb(data)

recreated_data = unpackb(msg_packet)

Using Datetime

from datetime import datetime as dtm
from dateutil import parser

def dtm_encode(obj):
    if isinstance(obj, dtm):
        return {
            '__dtm__': True,
            'obj_as_str': obj.isoformat()
        }
    return obj

def dtm_decode(obj):
    if '__dtm__' in obj:
        return parser.isoparse(obj['obj_as_str'])
    return obj

dat = {'a':5, 'dates': {'created_at': dtm.now()}}
serialized_dat = packb(dat, default=dtm_encode)
deserialzed_dat = unpackb(serialized_dat, object_hook=dtm_decode)

Why serialize the messages?

Serialization is the process of transforming an object into a format that it can be stored, at the destination the data can be deserialized to recreate the original object. Since Kafka uses the filesystem to store the messages, we need to serialize and deserialize the data. There are two protocols to serialize a message either by using Schema-based IDL(Interface Definition Language) or using Schemaless. While you can find numerous articles touching these protocols on the web, I have constituted a one-liner differentiating them.

Schemaless and Schema-based IDL

In Schema-based IDL, the primitives of the message are pre-defined where publishers and consumers can validate the message before working on it. On the other hand, schemaless can have custom primitives for every message.
If you're having a standard and rigid database schema, you might want to use Apache Avro which is the best fit and popularly used in combination with Apache Kafka. In contrast, if you're working with schemaless or bigdata oriented systems where primitives are highly unpredictable, there is an overhead of maintaining the Avro schemas, and the need for schemaless IDL arises.
These are some of the standard protocols most commonly chosen.

Protocol	Type
Thrift	Schema Based
Avro	Schema Based
JSON	Schemaless
MessagePack	Schemaless
BSON	Schemaless

I chose MessagePack among the list because it has an incredible performance, super simple to setup and start working with, and also much smaller and faster than JSON.

KafDrop

Kafdrop is an opensource web UI for viewing Kafka topics and browsing consumer groups. The tool displays information such as brokers, topics, partitions, consumers, and lets you view messages.

Example screen how a KafDrop looks like

Home Screen Home Screen not only covers the overview, but it also has information about the Brokers and topics available in the cluster

Topic Page Reveals all partitions info of the topic and renders the list of consumer groups subscribed to the topic

Features

List all topics of the cluster
Option to browse through a partition data of a topic
Tracks message offsets of all consumer groups subscribed to a topic
Create & Delete topics
Search and filter messages in a partition

Produce a Message and see in Kafdrop

from datetime import datetime as dtm

from msgpack import packb, unpackb
from confluent_kafka import Producer

'''
Note: Import or recreate "dtm_encode" function from the Code Snippet section 
'''


def delivery_report(err, m):
    if err is not None:
        print(f'Message -{m}, delivery failed: {err}')
    else:
        print(f'Message delivered to {m.topic()} [{m.partition()}]')


producer_config = {
    'bootstrap.servers': '172.16.6.9'
}
producer_topic = 'raghu-producer-test'
P = Producer(producer_config)
data = {
    'a':5,
    'created_at': dtm.now()
}
data_packet = packb(data, default=dtm_encode, use_bin_type=True)

P.produce(
    producer_topic,
    value=data_packet,
    callback=delivery_report,
)
P.poll(0.01)

Note: Refer code samples in my GitHub repo

Snapshot of Kafdrop

Best Practices

Before we close, I would like to mention some of the best practices

Compression reduces the packet size and helps to transmit much faster over the network, however, MessagePack shrinks the packet size by 30% which acts as compression and made me stay away from conventional compression libraries. Most standard workflows are designed using a compression mechanism, snappy & zlib are most widely accepted libraries for compression.
Encryption, which I more prefer than compression, is a good practice to ensure the data is secured over the network. It is not a tedious task to encrypt messages before pushing to Producer and decrypt at the Consumer end. There are lots of key-based encryption system which helps us to encrypt and decrypt the packets.
For my requirement, we have developed a thin wrapper on cryptography library to encrypt and decrypt messages using hybrid mechanism(Symmetric + Asymmetric), this functions like any other microservice to encrypt and decrypt the messages.
Message Headers, it is not an uncommon practice to incorporate headers in the message for several technical or architectural purposes. My team use headers to identify the version of the message, however, expect a new use-case with headers, refer the case to learn more.

Kafka - Playing with Consumer API using python library

RaghuNandan Neerukonda — Tue, 05 May 2020 16:57:21 +0000

Prerequisite: Requires prior knowledge(beginner to moderate level) on Kafka ecosystem.

Kafka has become one of the most widely used Message Broker for Event Bus architecture and Data Streams. We have been using Apache Kafka as a Message Broker for Microservices with CQRS design to build the services on different frameworks.

There are numerous articles available online which help developers to reuse the code snippets, however, it is mostly on Scala or Java. With this write-up, I would like to share some of the reusable code snippets for Kafka Consumer API using Python library confluent_kafka. confluent_kafka provides a good documentation explaining the funtionalities of all the API they support with the library. Their GitHub page also has adequate example codes.

Here, I would like to emphasize on two usecases which are rare but would definitely be used, at least a couple of times while working with message brokers.

Rewind Topic(Partition) offsets
Read from multiple partitions of different topics

Producer API

Firstly, lets get started with a sample code to produce a message.

import pickle
from confluent_kafka import Producer

my_dat = 'data that needs to be send to consumer'
P.produce(
    'my_topic',
    pickle.dumps(my_dat),
    callback=delivery_report,
)

pickle is used to serialize the data, this is not necessary if you working with integers and string, however, when working with timestamps and complex objects, we have to serialize the data.
Note: The best practise is to use Apache Avro, which is highly used in combination with Kafka.

Create a consumer and consume data

Initialize a consumer, subscribe to topics, poll consumer until data found and consume.

from confluent_kafka import Consumer

cfg = {
    'bootstrap.servers': '<kafka_host_ip>',
    'group.id': '<consumer_group_id>',
    'auto.offset.reset': 'earliest',
}
C = Consumer(cfg)
C.subscribe(['kafka-topic-1', 'kafka-topic-2', ])

for _ in range(10):
    msg = C.poll(0.05)
    if msg:
        dat = {
            'msg_value': msg.value(), # This is the actual content of the message
            'msg_headers': msg.headers(), # Headers of the message
            'msg_key': msg.key(), # Message Key
            'msg_partition': msg.partition(), # Partition id from which the message was extracted
            'msg_topic': msg.topic(), # Topic in which Producer posted the message to          
        }
        print(dat)

By default, consumer instances poll all the partitions of a topic, there is no need to poll each partition of topic to get the messages.
msg has a None value if poll method has no messages to return. Boolean check will help us to understand whether the poll to broker fetched message or not.
Valid message has not only data, it also has other functions which helps us to query or control the data.

Read from multiple partitions of different topics

Scenario:

Read from partition 1 of topic 1 starting with offset value 6
Read from partition 3 of topic 2 starting with offset value 5
Read from partition 2 of topic 1 starting with offset value 9

from confluent_kafka import Consumer, TopicPartition
cfg = {
    'bootstrap.servers': '172.16.18.187:9092',
    'group.id': 'hb-events-1',
    'auto.offset.reset': 'earliest',
}

C = Consumer(cfg)
C.assign(
    [
        TopicPartition(topic='kafka-topic-1', partition=1, offset=36),
        TopicPartition(topic='kafka-topic-2', partition=3, offset=35),
        TopicPartition(topic='kafka-topic-1', partition=2, offset=39),
    ]
)

no_msg_counter = 0
while True:
    msg = C.poll(0)
    if msg:
        no_msg_counter = 0
        dat = {
            'msg_val': msg.value(),
            'msg_partition': msg.partition(),
            'msg_topic': msg.topic()
        }
        print(dat)
    elif no_msg_counter > 10000:
        print('No Messages Found from a long time')
    else:
        no_msg_counter += 1
C.close()

When reading from a specific partition of a topic, assign is the best method to use instead of subscribe.
assign method accepts a list of TopicPartitions. TopicPartition is an instance which gets enrolled with one specific partition of a topic.

Rewind Topic(partition) offsets

Scenario:

Rewind the Partition 1 of topic-1 to offset 5
Reset the Partition 3 of topic-2

C = Consumer(cfg)
partition_topic_offsets = [
    TopicPartition('kafka-topic-1', partition=1, offset=5),
    TopicPartition('kafka-topic-2', partition=3, offset=0),
    ]
C.commit(offsets=partition_topic_offsets, async=False)
C.close()

C = Consumer(cfg)
C.subscribe(['kafka-topic-1', 'kafka-topic-2', ])

no_msg_counter = 0
while True:
    msg = C.poll(0.05)
    if msg:
        no_msg_counter = 0
        print(
            {
                'msg_value': msg.value(),
                'partition': msg.partition(),
                'headers': msg.headers(),
                'key': msg.key(),
                'offset': msg.offset(),
            }
        )
    elif no_msg_counter > 10000:
        break
    else:
        no_msg_counter += 1
C.close()

Irrespective of the current offset for the partition, we can rewind or reset the offset.
Reset or rewind offset values are set for a specific consumer groupid which was used to commit the offset, offsets of other consumer groups are unaffected

How we achieved this?

Create a list of TopicPartitions with the respective offset to reset
Commit these offsets to broker
When consumer subscribed to these topics poll, they get data from the recently set offset

Only retained messages are retrieved

Only message within the retention period are retrieved when you reset or rewind the offset.
If you lose or do not have a record of last successful offset, use OFFSET_BEGINNING, this will fetch data from the current beginning of the partition.

Use REST API

If you're frequently running out of issues and want to rewind, it is advised to periodically record/fetch the last successful offset to a table which look similar to events.
How frequent should we record?, depends on the business case. Recording every offset involves DB call which may slow down the service.
Create a wrapper REST-API which can update the table values.
Modify consumer groups to get last offset from table.