Racecar: a simple framework for Kafka consumers in Ruby
Racecar: a simple framework for Kafka consumers in Ruby
Racecar is a friendly and easy-to-approach Kafka consumer framework. It allows you to write small applications that process messages stored in Kafka topics while optionally integrating with your Rails models.
The framework is based on rdkafka-ruby, which, when used directly, can be a challenge: it's a flexible library with lots of knobs and options. Most users don't need that level of flexibility, though. Racecar provides a simple and intuitive way to build and configure Kafka consumers.
NOTE: Racecar requires Kafka 0.10 or higher.
Add this line to your application's Gemfile:
gem 'racecar'
And then execute:
$ bundle
Or install it yourself as:
$ gem install racecar
Then execute (if you're in a Rails application):
$ bundle exec rails generate racecar:install
This will add a config file in config/racecar.yml.
Racecar is built for simplicity of development and operation. First, a short introduction to the Kafka consumer concept as well as some basic background on Kafka.
Kafka stores messages in so-called partitions which are grouped into topics. Within a partition, each message gets a unique offset.
In Kafka, consumer groups are sets of processes that collaboratively process messages from one or more Kafka topics; they divide up the topic partitions amongst themselves and make sure to reassign the partitions held by any member of the group that happens to crash or otherwise becomes unavailable, thus minimizing the risk of disruption. A consumer in a group is responsible for keeping track of which messages in a partition it has processed – since messages are processed in-order within a single partition, this means keeping track of the offset into the partition that has been processed. Consumers periodically commit these offsets to the Kafka brokers, making sure that another consumer can resume from those positions if there is a crash.
A Racecar consumer is a simple Ruby class that inherits from Racecar::Consumer:
class UserBanConsumer { "Header-A" => 42, ... }
In order to avoid your consumer being kicked out of its group during long-running message processing operations, you'll need to let Kafka regularly know that the consumer is still healthy. There's two mechanisms in place to ensure that:
Heartbeats: They are automatically sent in the background and ensure the broker can still talk to the consumer. This will detect network splits, ungraceful shutdowns, etc.
Message Fetch Interval: Kafka expects the consumer to query for new messages within this time limit. This will detect situations with slow IO or the consumer being stuck in an infinite loop without making actual progress. This limit applies to a whole batch if you do batch processing. Use max_poll_interval to increase the default 5 minute timeout, or reduce batching with fetch_messages.
When a Racecar consumer shuts down, it gets the opportunity to tear down any resources held by the consumer instance. For example, it may make sense to close any open files or network connections. Doing so is simple: just implement a #teardown method in your consumer class and it will be called during the shutdown procedure.
class ArchiveConsumer e
puts "Failed to process message in #{message.topic}/#{message.partition} at offset #{message.offset}: #{e}"
# It's probably a good idea to report the exception to an exception tracker service.
end
Since the exception is handled by your #process method and is no longer raised, Racecar will consider the message successfully processed. Tracking these errors in an exception tracker or some other monitoring system is highly recommended, as you otherwise will have little insight into how many messages are being skipped this way.
If, on the other hand, the exception was cause by a temporary network or database problem, you will probably want to retry processing of the message after some time has passed. By default, if an exception is raised by the #process method, the consumer will pause all processing of the message's partition for some number of seconds, configured by setting the pause_timeout configuration variable. This allows the consumer to continue processing messages from other partitions that may not be impacted by the problem while still making sure to not drop the original message. Since messages in a single Kafka topic partition must be processed in order, it's not possible to keep processing other messages in that partition.
In addition to retrying the processing of messages, Racecar also allows defining an error handler callback that is invoked whenever an exception is raised by your #process method. This allows you to track and report errors to a monitoring system:
Racecar.config.on_error do |exception, info|
MyErrorTracker.report(exception, {
topic: info[:topic],
partition: info[:partition],
offset: info[:offset],
})
end
It is highly recommended that you set up an error handler. Please note that the info object contains different keys and values depending on whether you are using process or process_batch. See the instrumentation_payload object in the process and process_batch methods in the Runner class for the complete list.
A sample error might look like this:
E, [2022-10-09T11:28:29.976548 #15] ERROR -- : (try 5/10): Error for topic subscription #: Local: Not implemented (not_implemented)
Please see Compression
By default, Racecar will log to STDOUT. If you're using Rails, your application code will use whatever logger you've configured there.
In order to make Racecar log its own operations to a log file, set the logfile configuration variable or pass --log filename.log to the racecar command.
In order to gracefully shut down a Racecar consumer process, send it the SIGTERM signal. Most process supervisors such as Runit and Kubernetes send this signal when shutting down a process, so using those systems will make things easier.
In order to introspect the configuration of a consumer process, send it the SIGUSR1 signal. This will make Racecar print its configuration to the standard error file descriptor associated with the consumer process, so you'll need to know where that is written to.
In order to safely upgrade from Racecar v1 to v2, you need to completely shut down your consumer group before starting it up again with the v2 Racecar dependency.
Racecar v2 requires a C library (zlib) to compress the messages before producing to the topic. If not already installed on you consumer docker container, please install using following command in Dockerfile of consumer
apt-get update && apt-get install -y libzstd-dev
After checking out the repo, run bin/setup to install dependencies. Then, run rspec to run the tests. You can also run bin/console for an interactive prompt that will allow you to experiment.
The integration tests run against a Kafka instance that is not automatically started from within rspec. You can set one up using the provided docker-compose.yml by running docker compose up.
There can be behavioural inconsistencies between running the specs on your machine, and in the CI pipeline. Due to this, there is now a Dockerfile included in the project, which is based on the CircleCI ruby 2.7.8 image. This could easily be extended with more Dockerfiles to cover different Ruby versions if desired. In order to run the specs via Docker:
tests service from the docker-compose.ymldocker compose up -ddocker compose run --rm tests bundle exec rspecdocker compose run --rm tests bundle exec rspec spec/integration/consumer_spec.rbPlease note - your code directory is mounted as a volume, so you can make code changes without needing to rebuild
A new version is published to RubyGems.org every time a change to version.rb is pushed to the main branch.
In short, follow these steps:
version.rb,bundle lock to update Gemfile.lock,main, andTo create a pre-release from a non-main branch:
version.rb to something like 2.13.0.pre.1 or 3.0.0.beta.2,Bug reports and pull requests are welcome on GitHub. Feel free to join our Slack team and ask how best to contribute!
If you've discovered a bug, please file a Github issue, and make sure to include all the relevant information, including the version of Racecar, rdkafka-ruby, and Kafka that you're using.
If you have other questions, or would like to discuss best practises, or how to contribute to the project, join our Slack team!
Copyright 2017 Daniel Schierbeck & Zendesk
Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
No open issues yet, or sync has not completed.