resistant-kafka-avataa 0.9.8b15

High-level wrapper for confluent-kafka-python with built-in retry logic and protobuf support.

Resistant Kafka

Resistant Kafka is a Python library designed to simplify and stabilize interactions with Apache Kafka, both as a producer and a consumer.

Features

🔌 Easy integration into any Python service

To connect a consumer or producer, you just need to create one instance of the corresponding class: ConsumerInitializer or ProducerInitializer.

🔁 Serialisation | Deserialisation

The library allows you to serialize and deserialize data that you send using .proto formats. It also supports Schema Registry, which allows you to make sure that the data arrives in the correct format.

💾 Redis integration

We provide an easy way to connect to Redis. It is used for storing error messages and retrieving full information about each message.

🧾 Built-in logging of errors and events

Output of basic logs of connection of topic handlers in one consumer, as well as output of messages about successful sending of a message from the producer to certain topics.

🛡️ Resilience against consumer-side crashes

If an exception raises in the processor when reading a specific topic, by default, a detailed log about the dropped message will be issued and the consumer will continue its work.

In case you need to stop reading topic and raise exception - this option has also been added.

🧩 Handler creation for each topic in your service (Asynchronous)

One of the problems of working in the consumer is the case where the service reads several topics at the same time and this happens synchronously and in one handler.

We solved this problem!

By adding asynchronous reading of topics and adding the ability to read topics independently of each other. Even if one of them crashes (a crash will occur if you set the raise_error=True attribute in the kafka_processor) - the other handler will continue its work.

Also in this case it is very easy to separate the logic of processing messages of different topics if their keys, message type differ from each other.

Consumer Initializer

First Step. Add enviroments

Using the ConsumerConfig scheme you can configure the message reading handler in your service.

If reading of several topics is expected, then a more convenient way is to assemble common settings for connecting to Kafka and add them to the handler class (for example, to KafkaMessageProcessor) by **kwargs .

EXAMPLE:

from resistant_kafka.consumer_schemas import ConsumerConfig

process_task_1 = KafkaMessage1Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
)

process_task_2 = KafkaMessage2Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer2',
        processor_name='KafkaProcessor2',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
)

Second Step. Add processor

Processor is a class-handler of a specific topic. It allows to perform CRUD operations on received messages from a given topic independently of other processors.

⚠️ The name of the main method "process" is reserved and is required for installation.⚠️

⚠️Attribute "message" in main method "process" is required ⚠️

⚠️The decorator "kafka_processor" is also required ⚠️, which is responsible for the operation of the message stream and the stable operation of the main method "process". It has the attribute raise_error, which allows to raise an error, while the work of a specific handler will be stopped.

EXAMPLE:

from resistant_kafka.consumer import ConsumerInitializer, kafka_processor
from resistant_kafka_avataa.message_desirializers import MessageDeserializer

class KafkaMessage1Processor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig,
            deserializers: MessageDeserializer = None
    ):
        super().__init__(config=config, deserializers=deserializers)
        self._config = config
        self._deserializers = deserializers

    # required decorator
    # to raise error, instead logging @kafka_processor(raise_error=True)
    @kafka_processor()
    async def process(self, message):
        message_key = message.key().decode("utf-8")
        message_value = message.value().decode("utf-8")

        if message_value in ['WRONG_VALUE']:
            raise ValueError('You catch wrong value')
        
        # here your message proccessing

Third Step. Initialization

In order to start topic processors, you should use the "init_kafka_connection" method, to which you need to pass a list of instances of the processor-classes.

EXAMPLE:

from resistant_kafka.consumer import init_kafka_connection

process_task_1 = KafkaMessageProcessor1(
    config=ConsumerConfig(
        topic_to_subscribe='TOPIC_NAME_1',
        processor_name='KafkaMessageProcessor1',
        **consumer_config
    )
)

process_task_2 = KafkaMessageProcessor2(
    config=ConsumerConfig(
        topic_to_subscribe='TOPIC_NAME_2',
        processor_name='KafkaMessageProcessor2',
        **consumer_config
    )
)

init_kafka_connection(
    tasks=[process_task_1, process_task_2]
)

️⚠️In the way, where you have already created loop - use method "process_kafka_connection" ⚠️

import asyncio
from resistant_kafka.consumer import process_kafka_connection

asyncio.create_task(process_kafka_connection([inventory_changes_processor]))

Additional Step. Add security

To add security, you should set attribute security_config using class KafkaSecurityConfig

from resistant_kafka.common_schemas import KafkaSecurityConfig
from resistant_kafka.consumer_schemas import ConsumerConfig

security_config = KafkaSecurityConfig(
    oauth_cb=method_to_get_token,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)

consumer_config = ConsumerConfig(
        bootstrap_servers='HOST:PORT',
        group_id='CONSUMER_NAME',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        
        security_config=consumer_config
)

For mTLS/SSL connections, set optional certificate paths on KafkaSecurityConfig:

security_config = KafkaSecurityConfig(
    security_protocol='SSL',
    sasl_mechanisms='',
    ssl_ca_location='/path/to/ca.pem',
    ssl_certificate_location='/path/to/client.crt',
    ssl_key_location='/path/to/client.key',
    ssl_key_password='optional-key-password',
)

For SASL_SSL, combine SASL settings with the SSL fields above.

Additional Step. Add deserializers

In Kafka, messages are stored as bytes, so we need to get them in the format we expect. The library provides the ability to convert messages from bytes to objects using your .proto files or, if no .proto files are available, we will convert them to strings for your future processing of this data.

PROTO FILES

In case you have proto files that can help you format your messages - we can convert them from bytes to protobuf structure.

This can be done with Kafka Schema Registry, if your project doesn't have Kafka Schema Registry - we will convert bytes to strings.

You should use MessageDeserializer to registry your proto files with which you expect messages from the topic

REGISTRY .proto FILES

from resistant_kafka_avataa.message_desirializers import MessageDeserializer

deserializers_producer = MessageDeserializer(
    schema_registry_url="https://localhost:8081",
    topic='KafkaTesterProducer1'
)
deserializers_producer.register_protobuf_deserializer(ProtoFileWithData_1)
deserializers_producer.register_protobuf_deserializer(ProtoFileWithData_2)


process_task_1 = KafkaMessage1Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        **consumer_config
    ),
    deserializers=deserializers_producer_1
)

SIMPLE MESSAGES

from resistant_kafka_avataa.message_desirializers import MessageDeserializer

deserializers_producer_2 = MessageDeserializer(
    topic='KafkaTesterProducer2'
)

READ DESERIALIZED MESSAGES

Using the deserialize method you can convert the byte format of the message value from the format of objects described in .proto and end up with an object instead of bytes

class KafkaMessage1Processor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig,
            deserializers: MessageDeserializer = None
    ):
        super().__init__(config=config, deserializers=deserializers)
        self._config = config
        self._deserializers = deserializers

    @kafka_processor(store_error_messages=True)
    async def process(self, message):
        message_key = message.key().decode("utf-8")
        message_value = message.value().decode("utf-8")

        if message_value in ['WRONG_VALUE']:
            raise ValueError('You catch wrong value')
        
        deserialized_message = self._deserializers.deserialize(message=message)

Additional Step. Integrate Redis

    pip install redis=4.5.4

from resistant_kafka_avataa.common_schemas import RedisStoreConfig

redis_store_config = RedisStoreConfig(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
)

process_task = KafkaMessageProcessor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,

        redis_store_config=redis_store_config,
    )
)

class KafkaMessageProcessor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig
    ):
        super().__init__(config=config)
        self._config = config
    
    # IF "TRUE", ERROR MESSAGES DATA ARE STORED IN REDIS
    @kafka_processor(store_error_messages=True)
    async def process(self, message):
        
        # HERE YOU PROCESS KAFKA MESSAGES
        pass

Producer Initializer

First Step. Add enviroment variables

To configure a producer you will need only 2 fields: URL for connecting Kafka and the producer name.

producer_config = ProducerConfig(
        producer_name='KafkaTesterProducer1',
        bootstrap_servers='HOST:PORT',
)

Second Step. Add processor

The send_message method of ProducerInitializer class allows to send a message to a topic

Also an optional parameter of DataSend scheme is "headers" which allows to send additional information in a message without changing the structure of this message.

EXAMPLE:

task = ProducerInitializer(
    config=producer_config
)

task.send_message(
    data_to_send=DataSend(
        key='KEY1',
        value='VALUE1',
    )
)

# with headers
task.send_message(
    data_to_send=DataSend(
        key='KEY2',
        value='VALUE12,
        headers=[('additinal_key', 'additinal_value')]
    )
)

Additional Step. Add security

To add security, you should set attribute security_config using class KafkaSecurityConfig

from resistant_kafka import ProducerConfig
from resistant_kafka.common_schemas import KafkaSecurityConfig

security_config = KafkaSecurityConfig(
    oauth_cb=method_to_get_token,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)

producer_config = ProducerConfig(
        producer_name='KafkaTesterProducer1',
        bootstrap_servers='HOST:PORT',

        security_config=consumer_config
)

SSL/mTLS certificate paths can be set on KafkaSecurityConfig (see consumer security section above).

Additional Step. Add serializers

In Kafka, messages are stored as bytes, so we need to get them in the format we expect. The library provides the ability to convert messages from bytes to objects using your .proto files or, if no .proto files are available, we will convert them to strings for your future processing of this data.

PROTO FILES

In case you have proto files that can help you format your messages - we can convert them from bytes to protobuf structure.

This can be done with Kafka Schema Registry, if your project doesn't have Kafka Schema Registry - we will convert bytes to strings.

You should use MessageSerializer to registry your proto files with which you expect messages from the topic

REGISTRY .proto FILES

from resistant_kafka_avataa.message_serializers import MessageSerializer

serializer_task = MessageSerializer(
    schema_registry_url="https://localhost:8081",
    topic='KafkaTesterProducer1'
)
serializer_task.register_protobuf_deserializer(ProtoFile1)
serializer_task.register_protobuf_deserializer(ProtoFile2)

_producer_manager.send_message(
    data_to_send=DataSend(
        key=key,
        value=serializer_task.serialize(
            message_to_send=message_to_send,
            class_name=ProtoFile1
        ),
    ),
)

SIMPLE MESSAGES

from resistant_kafka_avataa.message_serializers import MessageSerializer

serializer_task = MessageSerializer(
    topic='KafkaTesterProducer1'
)
_producer_manager.send_message(
    data_to_send=DataSend(
        key=key,
        value=serializer_task.serialize(
            message_to_send=message_to_send
        ),
    ),
)

Installation

    pip install resistant-kafka-avataa

CONSUMER CODE EXAMPLE

from custom_utils import custom_token_method
from resistant_kafka.common_schemas import KafkaSecurityConfig
from resistant_kafka.consumer_schemas import ConsumerConfig
from resistant_kafka.consumer import ConsumerInitializer, kafka_processor, init_kafka_connection

consumer_config = KafkaSecurityConfig(
    oauth_cb=custom_token_method,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)


class KafkaMessage1Processor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig,
            deserializers: MessageDeserializer = None
    ):
        super().__init__(config=config, deserializers=deserializers)
        self._config = config
        self._deserializers = deserializers

    @kafka_processor(store_error_messages=True)
    async def process(self, message):
        message_key = message.key().decode("utf-8")
        message_value = message.value().decode("utf-8")

        if message_value in ['WRONG_VALUE']:
            raise ValueError('You catch wrong value')
        print('-----------------------------')
        print('KEY', message_key)
        print('VALUE', message_value)
        print('CONSUMER', self._config.topic_to_subscribe)
        print('-----------------------------')


class KafkaMessage2Processor(ConsumerInitializer):
    def __init__(
            self,
            config: ConsumerConfig,
            deserializers: MessageDeserializer = None
    ):
        super().__init__(config=config, deserializers=deserializers)
        self._config = config

    @kafka_processor()
    async def process(self, message):
        message_key = message.key().decode("utf-8")
        message_value = message.value().decode("utf-8")

        print('-----------------------------')
        print('KEY', message_key)
        print('VALUE', message_value)
        print('PRODUCER', self._config.topic_to_subscribe)
        print('-----------------------------')


deserializers_producer_1 = MessageDeserializer(
    schema_registry_url='http://localhost:8081',
    topic='KafkaTesterProducer1'
)
deserializers_producer_1.register_protobuf_deserializer(ProtoFileToDeserialize)

deserializers_producer_2 = MessageDeserializer(
    topic='KafkaTesterProducer2'
)

process_task_1 = KafkaMessage1Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer1',
        processor_name='KafkaProcessor1',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        redis_store_config=redis_store_config,
        security_config=consumer_config
    ),
    deserializers=deserializers_producer_1
)

process_task_2 = KafkaMessage2Processor(
    config=ConsumerConfig(
        topic_to_subscribe='KafkaTesterProducer2',
        processor_name='KafkaProcessor2',
        bootstrap_servers='localhost:9093',
        group_id='LocalTester1',
        auto_offset_reset='latest',
        enable_auto_commit=False,
        redis_store_config=redis_store_config,
        security_config=consumer_config
    ),
    deserializers=deserializers_producer_2
)

init_kafka_connection(
    tasks=[process_task_1, process_task_2]
)

PRODUCER CODE EXAMPLE

from custom_utils import custom_token_method
from resistant_kafka import ProducerInitializer, ProducerConfig, DataSend
from resistant_kafka.common_schemas import KafkaSecurityConfig

security_config = KafkaSecurityConfig(
    oauth_cb=custom_token_method,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanisms='OAUTHBEARER'
)

task = ProducerInitializer(
    config=ProducerConfig(
        producer_name='KafkaTesterProducer1',
        bootstrap_servers='HOST:PORT',
        security_config=security_config
    )
)
task.send_message(
    data_to_send=DataSend(
        key='KEY1',
        value='VALUE1',
    )
)
task.send_message(
    data_to_send=DataSend(
        key='KEY1',
        value='WRONG_VALUE'
    )
)

task = ProducerInitializer(
    config=ProducerConfig(
        producer_name='KafkaTesterProducer2',
        bootstrap_servers='HOST:PORT',
        security_config=security_config
    ),

)
task.send_message(
    data_to_send=DataSend(
        key='KEY2',
        value='VALUE2',
        headers=[
            ('key_1', 'value_1'),
            ('key_2', 'value_2'),
        ]
    ))