First Steps

Creating a simple Kafka consumer app

For our first demo we will create the simplest possible Kafka consumer and run it using ‘fastkafka run’ command.

The consumer will:

1. Connect to the Kafka Broker we setup in the Intro guide

2. Listen to the hello topic

3. Write any message received from the hello topic to stdout

To create the consumer, first, create a file named

hello_kafka_consumer.py and copy the following code to it:

from os import environ

from fastkafka import FastKafka
from pydantic import BaseModel, Field

kafka_server_url = environ["KAFKA_HOSTNAME"]
kafka_server_port = environ["KAFKA_PORT"]

kafka_brokers = {
    "localhost": {
        "description": "local development kafka",
        "url": kafka_server_url,
        "port": kafka_server_port
    }
}

class HelloKafkaMsg(BaseModel):
    msg: str = Field(
        ...,
        example="Hello",
        description="Demo hello world message",
    )

kafka_app = FastKafka(
    kafka_brokers=kafka_brokers
)
    
@kafka_app.consumes()
async def on_hello(msg: HelloKafkaMsg):
    print(f"Got data, msg={msg.msg}", flush=True)

!!! info "Kafka configuration"

This consumer script uses KAFKA_HOSTNAME and KAFKA_PORT environment vars, so make sure that you have exported them into your environment before running the following comand (e.g. in shell, for KAFKA_HOSTNAME, run: 'export KAFKA_HOSTNAME=kafka').

!!! warning "Remember to flush"

Notice the **flush=True** option when using print in our consumer. This is because standard python print function doesn't flush by default. To be able to log the worker outputs in real time when using fastkafka run command, the outputs need to be flushed, they will be logged when stopping the worker otherwise.

To run this consumer, in your terminal, run:

fastkafka run --num-workers=1 --kafka-broker localhost hello_kafka_consumer:kafka_app

After running the command, you should see something similar to the ouput below:

[INFO] fastkafka._components.test_dependencies: Java is already installed.
[INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...
[INFO] fastkafka._components.test_dependencies: Kafka is installed.
[INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...
[INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...
[INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...
[INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092
[878412]: [INFO] fastkafka._application.app: set_kafka_broker() : Setting bootstrap_servers value to '127.0.0.1:9092'
[878412]: [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...
[878412]: [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'bootstrap_servers': '127.0.0.1:9092', 'auto_offset_reset': 'earliest', 'max_poll_records': 100}
[878412]: [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.
[878412]: [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'hello'})
[878412]: [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'hello'}
[878412]: [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.
[878412]: [WARNING] aiokafka.cluster: Topic hello is not available during auto-create initialization
[878412]: [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'hello': 0}. 
Starting process cleanup, this may take a few seconds...
[INFO] fastkafka._server: terminate_asyncio_process(): Terminating the process 878412...
[878412]: [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.
[878412]: [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.
[INFO] fastkafka._server: terminate_asyncio_process(): Process 878412 terminated.

[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 877951...
[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 877951 terminated.
[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 877579...
[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 877579 terminated.

Now you can interact with your consumer, by sending the messages to the subscribed ‘hello’ topic, don’t worry, we will cover this in the next step of this guide.

Sending first message to your consumer

After we have created and run our first consumer, we should send a message to it, to make sure it is working properly.

If you are using the Kafka setup as described in the Intro guide, you can follow the steps listed here to send a message to the hello topic.

First, connect to your running kafka broker by running:

docker run -it kafka /bin/bash

Then, when connected to the container, run:

kafka-console-producer.sh --bootstrap-server=localhost:9092 --topic=hello

This will open an interactive connection to the hello topic, now you can write your mesages to the topic and they will be consumed by our consumer.

In the shell, type:

{"msg":"hello"}

and press enter. This will send a hello message to the topic which will be read by our running consumer and outputed to stdout.

Check the output of your consumer (terminal where you ran the ‘fastkafka run’ command) and confirm that your consumer has read the Kafka message. You shoud see something like this:

Got data, msg=hello

Creating a hello Kafka producer

Consuming messages is only a part of this Library functionality, the other big part is producing the messages. So, let’s create our first kafka producer which will send it’s greetings to our consumer periodically.

The producer will:

1. Connect to the Kafka Broker we setup in the Intro guide
2. Connect to the hello topic
3. Periodically send a message to the hello world topic

To create the producer, first, create a file named

hello_kafka_producer.py and copy the following code to it:

from os import environ

import asyncio
from pydantic import BaseModel, Field

from fastkafka import FastKafka
from fastkafka._components.logger import get_logger

kafka_server_url = environ["KAFKA_HOSTNAME"]
kafka_server_port = environ["KAFKA_PORT"]

kafka_brokers = {
    "localhost": {
        "description": "local development kafka",
        "url": kafka_server_url,
        "port": kafka_server_port
    }
}

class HelloKafkaMsg(BaseModel):
    msg: str = Field(
        ...,
        example="Hello",
        description="Demo hello world message",
    )

kafka_app = FastKafka(
    kafka_brokers=kafka_brokers
)

logger = get_logger(__name__)

@kafka_app.produces()
async def to_hello(msg: HelloKafkaMsg) -> HelloKafkaMsg:
    logger.info(f"Producing: {msg}")
    return msg

@kafka_app.run_in_background()
async def hello_every_second():
    while(True):
        await to_hello(HelloKafkaMsg(msg="hello"))
        await asyncio.sleep(1)

!!! info "Kafka configuration"

This producer script uses KAFKA_HOSTNAME and KAFKA_PORT environment vars, so make sure that you have exported them into your environment before running the following comand (e.g. in shell, for KAFKA_HOSTNAME, run: 'export KAFKA_HOSTNAME=kafka').

To run this producer, in your terminal, run:

fastkafka run --num-workers=1 --kafka-broker localhost hello_kafka_producer:kafka_app

After running the command, you should see something similar to the ouput below:

[INFO] fastkafka._components.test_dependencies: Java is already installed.
[INFO] fastkafka._components.test_dependencies: Kafka is installed.
[INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...
[INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...
[INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092
[879272]: [INFO] fastkafka._application.app: run_in_background() : Adding function 'hello_every_second' as background task
[879272]: [INFO] fastkafka._application.app: set_kafka_broker() : Setting bootstrap_servers value to '127.0.0.1:9092'
[879272]: [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:9092'}'
[879272]: [INFO] fastkafka._application.app: _populate_bg_tasks() : Starting background task 'hello_every_second'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [WARNING] aiokafka.cluster: Topic hello is not available during auto-create initialization
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
Starting process cleanup, this may take a few seconds...
[INFO] fastkafka._server: terminate_asyncio_process(): Terminating the process 879272...
[879272]: [INFO] hello_kafka_producer: Producing: msg='hello'
[879272]: [INFO] fastkafka._application.app: _shutdown_bg_tasks() : Cancelling background task 'hello_every_second'
[879272]: [INFO] fastkafka._application.app: _shutdown_bg_tasks() : Waiting for background task 'hello_every_second' to finish
[879272]: [INFO] fastkafka._application.app: _shutdown_bg_tasks() : Execution finished for background task 'hello_every_second'
[INFO] fastkafka._server: terminate_asyncio_process(): Process 879272 terminated.

[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 878808...
[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 878808 terminated.
[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 878435...
[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 878435 terminated.

Now, while the producer is running, it will send a HelloKafkaMsg every second to the hello kafka topic. If your consumer is still running, you should see the messages appear in its log.

Recap

In this guide we have:

1. Created a simple Kafka consumer using FastKafka
2. Sent a message to our consumer trough Kafka
3. Created a simple Kafka producer using FastKafka