Producer Management

It provides interfaces to start/stop producers, send messages to topics, and check producer running status. It also invokes functions to initialize and terminate producer instances to publish messages to Kafka, and it effectively checks the producer status to ensure that messages are sent successfully.

import json

from kafka_manager.kafka_manager import KafkaManager

bootstrap_servers = ['localhost:9092']  # Replace with your Kafka broker addresses
topic_name = 'example_topic'  # Replace topic name with your choice
group_id = 'example_group'  # Replace consumer group ID with your choice

# Create a KafkaManager instance
kafka_manager = KafkaManager(bootstrap_servers=bootstrap_servers)

# Start the Kafka producer
kafka_manager.start_producer();

# Send Kafka message
try:
    message_payload = json.dumps({
        "message_key": "message_value"
    })
    metadata = kafka_manager.send_message(topic=topic_name, value=message_payload)
    if metadata:
        print(f'Message sent successfully to Kafka topic: "{topic_name}"')
    else:
        print(f'Failed to send message to Kafka topic: "{topic_name}"')
except Exception as e:
    print(f'Error in sending message to Kafka topic: {e}')

# Stop Kafka producer
kafka_manager.stop_producer();

Consumer Management

It enables configuring various configurations to Create/Manage consumers and provides an interface to start/stop consumers. The Kafka Manager allows developers to create consumers per their application needs, such as different deserialization methods or offset management strategies. It provides a user-defined callback function to consume messages, allowing developers to define custom logic for processing each received message and enabling further data processing.

from kafka_manager.kafka_manager import KafkaManager

bootstrap_servers = ['localhost:9092']  # Replace with your Kafka broker addresses
topic_name = 'example_topic'  # Replace topic name with your choice
group_id = 'example_group'  # Replace consumer group ID with your choice

# Create a KafkaManager instance
kafka_manager = KafkaManager(bootstrap_servers=bootstrap_servers)

# Create a Kafka Consumer
consumer = kafka_manager.create_consumer(topics=[topic_name], group_id=group_id, auto_offset_reset='earliest')

# Start the Kafka Consumer
kafka_manager.start_consumer(consumer_id=group_id):

def message_handler(message):
    """
    This method is a callback function called by the consumer, which handles the received messages when a new message
    arrives.

    In production real-world application, the received message would be processed as follows:
    - Perform some business logic
    - Store the message in a database for further processing.
    - Message deserialization
    - etc.

    :param message: Message received from the consumer.
    """
    print(f'Received message: Partition={message.partition}, Offset={message.offset}, Value={message.value}')

# Consume Messages
kafka_manager.consume_messages(consumer_id=group_id, message_handler=message_handler)

Topic Management

Kafka Manager allows developers to create and delete topics dynamically, which serve as categories from which messages are published. It’s essential for managing data streams and evolving application requirements.

from kafka_manager.kafka_manager import KafkaManager

bootstrap_servers = ['localhost:9092']  # Replace with your Kafka broker addresses
topic_name = 'example_topic'  # Replace topic name with your choice
group_id = 'example_group'  # Replace consumer group ID with your choice

# Create a KafkaManager instance
kafka_manager = KafkaManager(bootstrap_servers=bootstrap_servers)

# For topic management connect to Kafka admin client
kafka_manager.connect_admin_client()

# Create a topic - (if it doesn't exist)
kafka_manager.create_topic(topic_name=topic_name, num_partitions=1, replication_factor=1)

Admin Client

It provides interfaces to connect to the Kafka Admin client and allows developers to perform administrative operations such as creating and deleting topics. However, the admin-client connection is vital to performing many advanced Kafka management tasks, such as describing cluster configurations and managing Kafka ACLs.

from kafka_manager.kafka_manager import KafkaManager

bootstrap_servers = ['localhost:9092']  # Replace with your Kafka broker addresses
topic_name = 'example_topic'  # Replace topic name with your choice
group_id = 'example_group'  # Replace consumer group ID with your choice

# Create a KafkaManager instance
kafka_manager = KafkaManager(bootstrap_servers=bootstrap_servers)

# Connect to Kafka admin client
admin_client = kafka_manager.connect_admin_client()

# Listing Consumer Groups
consumers_groups = admin_client.list_consumer_groups()
print(consumers_groups)

# Describing Consumer Groups
admin_client.describe_consumer_groups(list(consumers_groups))

Error Handling

To handle errors in Kafka due to network failures, broker failures, or misconfigurations, Kafka Manager handles these exceptions efficiently and ensures application stability.

Resource Management

Kafka Manager resource management ensures that all connections to Kafka are correctly closed. It provides a close() function for proper shutdown, which prevents resource leaks and potential data corruption. It’s essential for maintaining data integrity and managing the Kafka cluster and application.

License

MIT License, See LICENSE.