Note that the component holds a reference to kafka::Producer - a client to the Kafka (its writer). That client is thread safe, you can use it concurrently from different threads and tasks.
To use producer in the program, include it with
#include <userver/kafka/producer.hpp>
Produce messages to topic
To send messages to topics, let us declare and implement function kafka_sample::Produce that accepts a producer reference and the kafka_sample::RequestMessage and uses kafka::Producer::Send method to schedule its send to the given topic and asynchronously wait its delivery:
// Subscriptions must be the last fields! Add new fields above this comment.
kafka::ConsumerScope consumer_;
};
Note that component holds a reference to kafka::ConsumerScope - a client to the Kafka (its reader). Consumer should be launched on the component start to asynchronously handle consumed messages.
To use consumer in the program, include it with
#include <userver/kafka/consumer_scope.hpp>
Consume messages from topics
We should implement a method or function that handles the messages consumed from topics. Here, we are reading the message batch and call the testpoint message_consumed on each message key. But in real service, the process may take a long time and do worthy things.
At first, we should find a consumer instance and save it in our component's field. After that we start the asynchronous message processing process, passing a callback to ConsumerScope::Start that calls the kafka_sample::Consume and commits the messages offsets on success (no exception thrown in Consume).
Static configuration of service is quite close to the configuration from Writing your first HTTP server except for the handlers and producer and consumer settings.
Consumer settings are:
kafka-consumer:# Kafka consumer component name must start with kafka-consumergroup_id:test-group# Consumers group current consumer will join on starttopics:# List of topics consumer starts to listen-test-topic-1-test-topic-2auto_offset_reset:smallest# Strategy for reading topics without committed offsetsmax_batch_size:5# Number of messages consumers waits until user-callback is calledsecurity_protocol:PLAINTEXT# Broker connection settings
And basic producer setting is:
kafka-producer:# Kafka producer component name must start with kafka-producerdelivery_timeout:10s# Timeout for message deliveryqueue_buffering_max:100ms# Time, producer waits before sending a messageenable_idempotence:true# Whether to enable idempotence modesecurity_protocol:PLAINTEXT# Broker connection settings
Secdist
Kafka clients need to know the Kafka brokers endpoints and, optionally, the connection username and password.
Producer and Consumer read the aforementioned settings from Secdist in format:
"username": "SASL2 username (may be empty if use PLAINTEXT)",
"password": "SASL2 password (may be empty if use PLAINTEXT)"
}
}
}
Build and run
To build the sample, execute the following build steps at the userver root directory:
mkdir build_release
cd build_release
cmake -DCMAKE_BUILD_TYPE=Release ..
make userver-samples-kafka_service
The sample could be started by running make start-userver-samples-kafka_service. The command would invoke testsuite start target that sets proper paths in the configuration files, prepares and starts the Kafka broker, and starts the service.
Make sure to place secdist settings into /etc/kafka_service/secure_data.json and define the KAFKA_HOME environment variable with the path to Kafka binaries.
Also set the TESTSUITE_KAFKA_CUSTOM_TOPICS environment to create topics on server start.
For unit testing your service logic, it is useful to add functions that accepts the producer or consumer and call them both from service code and from tests.
To create producer and consumer instances in userver unit tests you can use kafka::utest::KafkaCluster fixture. Inherit your testing class from it and use UTEST_F macros.
Yandex Taxi Testsuite framework has Kafka plugin that helps set environment for functional service testing. Also, userver has some custom fixtures in pytest_userver.plugins.kafka
Fixtures and environment settings settings may be included in your conftest.py with:
pytest_plugins = ['pytest_userver.plugins.kafka']
Also, you can redefine a service_env fixture and pass kafka_secdist fixture to generate secdists settings for all yours consumers and producers:
@pytest.fixture(scope='session')
def kafka_components() -> List[str]:
return ['kafka-consumer', 'kafka-producer']
@pytest.fixture(scope='session')
def service_env(kafka_secdist) -> dict:
"""
Note: kafka_secist fixture generates the secdist config
