FAQ
- Does Karafka require Ruby on Rails?
- Why there used to be an ApplicationController mentioned in the Wiki and some articles?
- Does Karafka require Redis and/or Sidekiq to work?
- Could an HTTP controller also consume a fetched message through the Karafka router?
- Does Karafka require a separate process running?
- Can I start Karafka process with only particular consumer groups running for given topics?
- Can I use
#seek
to start processing topics partition from a certain point? - Why Karafka does not pre-initializes consumers prior to first message from a given topic being received?
- Does Karafka restart dead PG connections?
- Does Karafka require gems to be thread-safe?
- When Karafka is loaded via railtie in test env, SimpleCov does not track code changes
- Can I use Thread.current to store data in between batches?
- Why Karafka process does not pick up newly created topics until restarted?
- Why is Karafka not doing work in parallel when I started two processes?
- Can I remove a topic while the Karafka server is running?
- What is a forceful Karafka stop?
- Can I use AWS MSK Serverless with IAM authentication?
- Why can't I connect to Kafka from another Docker container?
- How can I configure multiple bootstrap servers?
- Why, when using
cooperative-sticky
rebalance strategy, all topics get revoked on rebalance? - What will happen with uncommitted offsets during a rebalance?
- Can I use Karafka with Ruby on Rails as a part of an internal gem?
- Can I skip messages on errors?
- What does static consumer fenced by other consumer with same group.instance.id mean?
- Why, in the Long-Running Jobs case,
#revoked
is executed even if#consume
did not run because of revocation? - Why am I seeing
Rdkafka::RdkafkaError (Local: Timed out (timed_out)
error when producing larger quantities of messages? - Do I need to use
#revoked?
when not using Long-Running jobs? - Can I consume from more than one Kafka cluster at the same time?
- Why Karafka uses
karafka-rdkafka
instead ofrdkafka
directly? - Why am I seeing an
Implement this in a subclass
error? - What is Karafka
client_id
used for? - How can I increase Kafka and Karafka max message size?
- Why do DLQ messages in my system keep disappearing?
- What is the optimal number of threads to use?
- Can I use several producers with different configurations with Karafka?
- What is the Unsupported value "SSL" for configuration property "security.protocol": OpenSSL not available at build time?
Does Karafka require Ruby on Rails?
No. Karafka is a fully independent framework that can operate in a standalone mode. It can be easily integrated with any Ruby-based application, including those written with Ruby on Rails. Please follow the Integrating with Ruby on Rails and other frameworks Wiki section.
Why there used to be an ApplicationController mentioned in the Wiki and some articles?
You can name the main application consumer with any name. You can even call it ApplicationController
or anything else you want. Karafka will sort that out, as long as your root application consumer inherits from the Karafka::BaseConsumer
. It's not related to Ruby on Rails controllers. Karafka framework used to use the *Controller
naming convention up until Karafka 1.2 where it was changed because many people had problems with name collisions.
Does Karafka require Redis and/or Sidekiq to work?
No. Karafka is a standalone framework, with an additional process that will be used to consume Kafka messages.
Could an HTTP controller also consume a fetched message through the Karafka router?
No. Kafka messages can be consumed only using Karafka consumers. You cannot use your Ruby on Rails HTTP consumers to consume Kafka messages, as Karafka is not an HTTP Kafka proxy. Karafka uses Kafka API for messages consumption.
Does Karafka require a separate process running?
No, however, it is recommended. By default, Karafka requires a separate process (Karafka server) to consume and process messages. You can read about it in the Consuming messages section of the Wiki.
Karafka can also be embedded within another process so you do not need to run a separate process. You can read about it here.
Can I start Karafka process with only particular consumer groups running for given topics?
Yes. Karafka allows you to listen with a single consumer group on multiple topics, which means that you can tune up the number of threads that Karafka server runs, accordingly to your needs. You can also run multiple Karafka instances, specifying consumer groups that should be running per each process using the --include-consumer-groups
server flag as follows:
bundle exec karafka server --include-consumer-groups group_name1 group_name3
You can also exclude particular groups the same way:
bundle exec karafka server --exclude-consumer-groups group_name1 group_name3
Visit the CLI section of our docs to learn more about how to limit the scope of things to which the server subscribes.
Can I use #seek
to start processing topics partition from a certain point?
Karafka has a #seek
consumer method that can be used to do that.
Why Karafka does not pre-initializes consumers prior to first message from a given topic being received?
Because Karafka does not have knowledge about the whole topology of a given Kafka cluster. We work on what we receive dynamically building consumers when it is required.
Does Karafka restart dead PG connections?
Karafka starting from 2.0.16
will automatically clean dead ActiveRecord connections. No extra action is needed.
Does Karafka require gems to be thread-safe?
Yes. Karafka uses multiple threads to process data, similar to how Puma or Sidekiq does it. The same rules apply.
When Karafka is loaded via a railtie in test env, SimpleCov does not track code changes
Karafka hooks with railtie to load karafka.rb
. Simplecov needs to be required before any code is loaded.
Can I use Thread.current to store data between batches?
No. The first available thread will pick up work from the queue to better distribute work. This means that you should not use Thread.current
for any type of data storage.
Why Karafka process does not pick up newly created topics until restarted?
- Karafka in the
development
mode will refresh cluster metadata every 5 seconds. It means that it will detect topic changes fairly fast. - Karafka in
production
will refresh cluster metadata every 5 minutes. It is recommended to create production topics before running consumers.
The frequency of cluster metadata refreshes can be changed via topic.metadata.refresh.interval.ms
in the kafka
config section.
Why is Karafka not doing work in parallel when I started two processes?
Please make sure your topic contains more than one partition. Only then Karafka can distribute the work to more processes. Keep in mind, that all the topics create automatically with the first message sent will always contain only one partition. Use the Admin API to create topics with more partitions.
Can I remove a topic while the Karafka server is running?
Not recommended. You may encounter the following errors if you decide to do so:
ERROR -- : librdkafka internal error occurred: Local: Unknown partition (unknown_partition)
ERROR -- :
INFO -- : rdkafka: [thrd:main]: Topic extractor partition count changed from 1 to 0
ERROR -- : librdkafka internal error occurred: Broker: Unknown topic or partition (unknown_topic_or_part)
It is recommended to stop Karafka server instances and then remove and recreate the topic.
What is a forceful Karafka stop?
When you attempt to stop Karafka, you may notice the following information in your logs:
Received SIGINT system signal
Stopping Karafka server
Forceful Karafka server stop
When you ask Karafka to stop, it will wait for all the currently running jobs to finish. The shutdown_timeout
configuration setting limits the time it waits. After this time passes and any work in listeners or workers are still being performed, Karafka will attempt to forcefully close itself, stopping all the work in the middle. If you see it happen, it means you need to either:
- extend the
shutdown_timeout
value to match your processing patterns - debug your code to check what is causing the extensive processing beyond the
shutdown_timeout
In any case, it is not recommended to ignore this if it happens frequently.
Can I use AWS MSK Serverless with IAM authentication?
No. IAM is a custom authentication engine that is not a part of the Kafka protocol and is not supported by librdkafka
.
Karafka supports the standard SASL + SSL mechanisms available for MSK. You can read more about it here.
Why can't I connect to Kafka from another Docker container?
You need to modify the docker-compose.yml
KAFKA_ADVERTISED_HOST_NAME
value. You can read more about it here.
How can I configure multiple bootstrap servers?
You need to define them comma-separated under kafka
bootstrap.servers
configuration key:
class KarafkaApp < Karafka::App
setup do |config|
config.client_id = 'my_application'
# This value needs to be a string string with comma separated servers
config.kafka = {
'bootstrap.servers': 'server1.address:9092, server2.address:9092'
}
end
end
Why, when using cooperative-sticky
rebalance strategy, all topics get revoked on rebalance?
This behavior can occur if you are using blocking mark_as_consumed!
method and the offsets commit happens during rebalance. When using cooperative-sticky
we recommend using mark_as_consumed
instead.
What will happen with uncommitted offsets during a rebalance?
When using mark_as_consumed
, offsets are stored locally and periodically flushed to Kafka asynchronously.
Upon rebalance, all uncommitted offsets will be committed before a given partition is re-assigned.
Can I use Karafka with Ruby on Rails as a part of an internal gem?
Karafka 2.0 has Rails auto-detection, and it is loaded early, so some components may be available later, e.g., when ApplicationConsumer inherits from BaseConsumer that is provided by the separate gem that needs an initializer.
Moreover, despite the same code base, some processes (rails s
, rails db:migrate
, sidekiq s
) may not need to know about karafka, and there is no need to load it.
The problem is presented in this example app PR.
To mitigate this, you can create an empty karafka bootfile. With a file structure like this:
+-- karafka_root_dir
| +-- karafka.rb # default bootfile (empty file)
| +-- karafka_app.rb # real bootfile with Karafka::App definition and other stuff
| +-- ...
It is possible to postpone the definition of the Karafka app and do it manually whenever & wherever the user wants (karafka_app.rb
could be loaded for example, in some initializer).
# karafka_app.rb
class KarafkaApp < Karafka::App
setup do |config|
config.client_id = 'my_application'
...
end
end
# config/initializers/karafka_init.rb
require 'karafka_root_dir/karafka_app'
Still not a perfect solution because karafka gem is still loaded.
Note: This description was prepared by AleksanderSzyszka.
Can I skip messages on errors?
Karafka Pro can skip messages non-recoverable upon errors as a part of the Enhanced Dead Letter Queue feature. You can read about this ability here.
What does static consumer fenced by other consumer with same group.instance.id mean?
If you see such messages in your logs:
Fatal error: Broker: Static consumer fenced by other consumer with same group.instance.id
It can mean two things:
- You are using the Karafka version before
2.0.20
. If that is the case, please upgrade. - Your
group.instance.id
is not unique within your consumer group. You must always ensure that the value you assign togroup.instance.id
is unique within the whole consumer group, not unique per process or machine.
Why, in the Long-Running Jobs case, #revoked
is executed even if #consume
did not run because of revocation?
The #revoked
will be executed even though the #consume
did not run upon revocation because #revoked
can be used to teardown resources initialized prop to #consume
. For example, for things initialized in a custom initialize
method.
Why am I seeing Rdkafka::RdkafkaError (Local: Timed out (timed_out)
error when producing larger quantities of messages?
If you are seeing following error:
Rdkafka::RdkafkaError (Local: Timed out (timed_out)
It may mean one of four things:
- High probability: Broker can't keep up with the produce rate.
- High probability if you use
partition_key
: Broker is temporarily overloaded and cannot return info about the topic structure. A retry mechanism has been implemented in WaterDrop2.4.4
to mitigate this. - Low probability: Slow network connection.
- Low probability: SSL configuration issue. In this case, no messages would reach the broker.
WaterDrop dispatches messages to librdkafka
and librdkafka
constructs message sets out of it. By default, it does it every five milliseconds. If you are producing messages fast, it may become inefficient for Kafka because it has to deal with separate incoming message sets and needs to keep up. Please consider increasing the queue.buffering.max.ms
, so the batches are constructed less often and are bigger.
Additionally, you may also:
- Dispatch smaller batches using
#produce_many_sync
.Effectively it will throttle the process that way. - Establish a limit on how many messages you want to dispatch at once. This will prevent you from scenarios where you accidentally flush too much. If you dispatch based on an array of samples, you can do it that way:
data_to_dispatch.each_slice(2_00) do |data_slice|
Karafka.producer.produce_many_sync(data_slice)
end
Do I need to check #revoked?
when not using Long-Running jobs?
In a stable system, no. The Karafka default offset management strategy should be more than enough. It ensures that after batch processing as well as upon rebalances, before partition reassignment, all the offsets are committed.
You can read about Karafka's revocation/rebalance behaviors here and here.
Can I consume from more than one Kafka cluster simultaneously?
Yes. Karafka allows you to redefine kafka
settings on a per-topic basis. You can create separate consumer groups to consume from separate clusters:
class KarafkaApp < Karafka::App
setup do |config|
# ...
end
routes.draw do
consumer_group :group_name do
topic :example do
kafka('bootstrap.servers': 'cluster1:9092')
consumer ExampleConsumer
end
topic :example2 do
kafka('bootstrap.servers': 'cluster1:9092')
consumer ExampleConsumer2
end
end
consumer_group :group_name2 do
topic :example3 do
kafka('bootstrap.servers': 'cluster2:9092')
consumer Example2Consumer3
end
end
end
end
Why Karafka uses karafka-rdkafka
instead of rdkafka
directly?
We release our version of the rdkafka
gem to ensure it meets our quality and stability standards. That way, we ensure that unexpected rdkafka
releases will not break the Karafka ecosystem.
Why am I seeing an Implement this in a subclass
error?
[bc01b9e1535f] Consume job for ExampleConsumer on my_topic started
Worker processing failed due to an error: Implement this in a subclass
This error occurs when you have defined your consumer but without a #consume
method:
BAD:
class ExampleConsumer < Karafka::BaseConsumer
# No consumption method
end
GOOD:
class ExampleConsumer < Karafka::BaseConsumer
def consume
messages.each do |message|
puts message.payload
end
end
end
What is Karafka client_id
used for?
Karafka client_id
is, by default, used for two things:
- Building ids for consumer groups using the default consumer mapper.
- Populating kafka
client.id
value.
kafka client.id
is a string passed to the server when making requests. This is to track the source of requests beyond just IP/port by allowing a logical application name to be included in server-side request logging.
Therefore the client_id
should be shared across multiple instances in a cluster or horizontally scaled application but distinct for each application.
How can I increase Kafka and Karafka max message size?
To make Kafka accept messages bigger than 1MB, you must change both Kafka and Karafka configurations.
To increase the maximum accepted payload size in Kafka, you can adjust the message.max.bytes
and replica.fetch.max.bytes
configuration parameters in the server.properties file. These parameters controls the maximum size of a message the Kafka broker will accept.
To allow WaterDrop (Karafka producer) to send bigger messages, you need to:
- set the
max_payload_size
config option to value in bytes matching your maximum expected payload. - set
kafka
scopedmessage.max.bytes
to the same value.
You can do this by reconfiguring WaterDrop during Karafka setup:
class KarafkaApp < Karafka::App
setup do |config|
config.producer = ::WaterDrop::Producer.new do |producer_config|
# Use all the settings already defined for consumer by default
producer_config.kafka = ::Karafka::Setup::AttributesMap.producer(config.kafka.dup)
producer_config.logger = config.logger
# Alter things you want to alter
producer_config.max_payload_size = 1_000_000_000
producer_config.kafka[:'message.max.bytes'] = 1_000_000_000
end
end
end
It is essential to keep in mind that increasing the maximum payload size may impact the performance of your Kafka cluster, so you should carefully consider the trade-offs before making any changes.
Note: If you do not allow bigger payloads and try to send them, you will end up with one of the following errors:
WaterDrop::Errors::MessageInvalidError {:payload=>"is more than `max_payload_size` config value"}
or
Rdkafka::RdkafkaError (Broker: Message size too large (msg_size_too_large)):
Why do DLQ messages in my system keep disappearing?
DLQ messages may disappear due to many reasons. Some possible causes include the following:
- The DLQ topic has a retention policy that causes them to expire and be deleted.
- The DLQ topic is a compacted topic, which only retains the last message with a given key.
- The messages are being produced to a DLQ topic with a replication factor of 1, which means that if the broker storing the messages goes down, the messages will be lost.
For more details, please look at the Compacting limitations section of the DLQ documentation.
What is the optimal number of threads to use?
The optimal number of threads for a specific application depends on various factors, including the number of processors and cores available, the amount of memory available, and the particular tasks the application performs and their type. In general, increasing number of threads brings the most significant benefits for IO-bound operations.
It's recommended to use the number of available cores to determine the optimal number of threads for an application.
When working with Karafka, you also need to take into consideration things that may reduce the number of threads being in use, that is:
- Your topics count.
- Your partitions count.
- Number of processes within a given consumer group.
- To how many topics and partitions a particular process is subscribed to.
Karafka can parallelize work in a couple of scenarios, but unless you are a Karafka Pro user and you use Virtual Partitions, in a scenario where your process is assigned to a single topic partition, the work will always happen only in a single thread.
You can read more about Karafka and Karafka Pro concurrency model here.
It's also essential to monitor the performance of the application and the system as a whole while experimenting with different thread counts. This can help you identify bottlenecks and determine the optimal number of threads for the specific use case.
Remember that the optimal number of threads may change as the workload and system resources change over time.
Can I use several producers with different configurations with Karafka?
Yes. You can create as many producers as you want using WaterDrop API directly:
producer = WaterDrop::Producer.new do |config|
config.deliver = true
config.kafka = {
'bootstrap.servers': 'localhost:9092',
'request.required.acks': 1
}
end
and you can use them.
There are a few things to keep in mind, though:
- Producers should be long-lived.
- Producers should be closed before the process shutdown to ensure proper resource finalization.
- You need to instrument each producer using the WaterDrop instrumentation API.
- Karafka itself uses the
Karafka#producer
internal reasons such as error tracking, DLQ dispatches, and more. This means that the default producer instance should be configured to operate within the scope of Karafka's internal functionalities.
What is the Unsupported value "SSL" for configuration property "security.protocol": OpenSSL not available at build time?
If you are seeing the following error:
`validate!':
{:kafka=>"Unsupported value "SSL" for configuration property "security.protocol":
OpenSSL not available at build time"} (Karafka::Errors::InvalidConfigurationError)
It means you want to use SSL, but librdkafka
was built without it. You have to:
- Uninstal it by running
gem remove karafka-rdkafka
- Install
openssl
(OS dependant but for macos, that would bebrew install openssl
) - Run
bundle install
again, solibrdkafka
is recompiled with SSL support.