tamarco-kafka
Kafka resource for Tamarco microservice framework. It runs a confluent-kafka client in a thread.
This repository is a plugin for Tamarco, for more information go to Tamarco main repository.
Settings
This resource depends on the following configuration:
system:
resources:
kafka:
bootstrap_servers: kafka:9092The bootstrap servers are the address of the members of a kafka cluster separated by coma.
Input and outputs
The inputs and outputs need to be declared in the resource.
Input
The input can be used with two different patterns, as decorator and as async stream.
This resource only supports balanced consumer groups with auto commit.
Async stream
This usage case uses the input as asynchronous iterator to consume the metric stream.
class MyMicroservice(Microservice):
name = "input_example"
metrics_input = KafkaInput(topic='metrics', codec=JsonCodec)
kafka = KafkaResource(inputs=[metrics_input])
@task
async def metrics_consumer(self):
async for metric in self.metrics_input:
self.logger.info(f'Consumed message from metrics topic: {metric}')Decorator
This usage case declares a function as handler of the messages, and the resource is going to open automatically a coroutine to consume each message.
class MyMicroservice(Microservice):
name = "input_example"
kafka = KafkaResource(inputs=[metrics_input])
@KafkaInput(resource=kafka, topic='metrics', codec=JsonCodec)
async def metrics_handler(self, message):
self.logger.info(f'Consumed message from metrics topic: {message}')Output
It is a Kafka producer very simple to use.
class MyMicroservice(Microservice):
name = "output_example"
metrics_output = KafkaOutput(topic='metrics', codec=JsonCodec)
kafka = KafkaResource(outputs=[metrics_output])
@task_timer(interval=1000, autostart=True)
async def metrics_producer(self):
metrics_message = {'metrics': {'cat': 'MEOW'}}
await self.metrics_output.push(metrics_message)
self.logger.info(f'Produced message {metrics_message} to metrics topic')How to run the examples
To run them you just need to launch the docker-compose, install the requirements and run it.
pip install -r examples/requirements.txt
docker-compose up -d
python examples/microservice_stream_input.py
