v2.5.0

Author: mensfeld

Advanced/Admin-API.md

@@ -389,7 +389,7 @@ end
 
 ## Renaming a Consumer Group
 
-!!! Warning "Never rename active consumer groups"
+!!! Warning "Never Rename Active Consumer Groups"
 
     This method should **not** be used on actively running consumer groups, as it involves creating a temporary consumer to handle offset migration. Running this operation on active groups may cause unexpected behavior.
 
@@ -409,9 +409,41 @@ When using `rename_consumer_group`, the method ensures that offsets from the old
 
     If the new consumer group already exists, the offsets from the old group will be merged into it. This may result in the continuation of message processing from the combined offsets, so plan accordingly.
 
+## Copying a Consumer Group
+
+!!! warning "Never Copy Active Consumer Groups"
+
+  This method should **not** be used on actively running consumer groups, as it involves creating a temporary consumer to handle offset migration. Running this operation on active groups may cause unexpected behavior.
+
+The `#copy_consumer_group` method in Karafka Admin API allows you to copy offsets from an existing consumer group to another while preserving its consumption state for specific topics. This functionality is useful when creating a duplicate consumer group with the same consumption progress as an existing one.
+
+```ruby
+Karafka::Admin.copy_consumer_group(
+  'source_group_name',
+  'target_group_name',
+  ['topic1', 'topic2']
+)
+```
+
+When using `#copy_consumer_group`, the method ensures that offsets from the source consumer group are transferred to the target one, maintaining continuity in message consumption. You need to specify which topics should have their offsets copied during the process, giving you control over what gets migrated.
+
+!!! Tip "Offset Merger with Existing Consumer Groups"
+
+    If the target consumer group already exists, the offsets from the source group will be merged into it. This may result in the continuation of message processing from the combined offsets, so plan accordingly.
+
+The method returns `true` if offsets were successfully copied or `false` if there was nothing to copy (for example, if the source consumer group doesn't exist or has no committed offsets for the specified topics).
+
+This functionality is particularly useful for:
+
+- Creating backup consumer groups before making significant changes
+- Testing new consumer configurations with the same consumption progress
+- Setting up disaster recovery scenarios
+
+Unlike `#rename_consumer_group`, this method preserves the source consumer group, allowing both groups to exist simultaneously.
+
 ## Deleting a Consumer Group
 
-!!! warning "Never delete active consumer groups"
+!!! warning "Never Delete Active Consumer Groups"
 
     This method should only be used for consumer groups **not** actively used. Deleting a consumer group that is currently in use (running) can lead to data loss, inconsistencies, or unexpected behavior in your Kafka cluster.
 
@@ -425,7 +457,7 @@ Karafka::Admin.delete_consumer_group('your_consumer_group_name')
 
 ## Changing an Offset of a Consumer Group
 
-!!! warning "Never alter active consumer groups"
+!!! warning "Never Alter Active Consumer Groups"
 
     This method should only be used for consumer groups **not** actively used. Altering a consumer group that is currently in use (running) can lead to data loss, inconsistencies, or unexpected behavior in your Kafka cluster.
 

Advanced/Concurrency-and-Multithreading.md

@@ -181,3 +181,24 @@ Karafka provides an advanced operation mode known as Swarm, designed to optimize
 In Swarm Mode, Karafka forks multiple independent processes, each capable of running concurrently. This approach allows the framework to manage and supervise these processes effectively, ensuring high availability and resilience. By doing so, Karafka can better distribute the workload across available CPU cores, minimizing bottlenecks and maximizing processing speed.
 
 Swarm has its own section. You can read about it [here](Swarm-Multi-Process).
+
+## Setting Thread Priority
+
+Karafka supports explicit thread priority configuration. Adjusting thread priorities can mitigate performance issues caused by mixed workloads, particularly by reducing latency when running IO-bound and CPU-bound tasks concurrently.
+
+Karafka processing threads have a default priority set to `-1`. Lowering this priority further can significantly reduce tail latency for IO-bound tasks, ensuring more balanced resource allocation, especially in scenarios with CPU-intensive workloads that could monopolize the Global VM Lock (GVL).
+
+```ruby
+class KarafkaApp < Karafka::App
+  setup do |config|
+    # Lower worker thread priority to prevent CPU-bound tasks from starving IO-bound threads
+    config.worker_thread_priority = -3
+  end
+end
+```
+
+Lowering thread priority (e.g., negative values like `-1`, `-3`) can significantly reduce tail latency for IO-bound tasks. This ensures more balanced resource allocation, especially in scenarios with CPU-intensive workloads that could monopolize the Global VM Lock (GVL).
+
+!!! tip "Thread Priority and GVL"
+
+    Ruby employs a Global VM Lock (GVL) that ensures only one thread executes Ruby code at a time. The Ruby VM switches threads roughly every 100ms (thread quantum) unless explicitly released (such as during IO operations). CPU-intensive tasks holding the GVL for the entire quantum period can significantly increase latency for other threads, especially those performing quick IO tasks. Adjusting thread priority mitigates this issue by influencing the scheduling decisions and allowing shorter, IO-bound threads more frequent access to the CPU.

Consuming-Messages.md

@@ -348,7 +348,7 @@ iterator = ::Karafka::Pro::Iterator.new(
 )
 
 iterator.each do |message|
-  # Cast to integer because headers are always string
+  # Cast to integer because headers are always strings or arrays of strings
   next unless message.headers['user-id'].to_i == 5
 
   user_5_events << message

Pro/Enhanced-Dead-Letter-Queue.md

@@ -356,3 +356,37 @@ When implementing a custom DLQ strategy in Karafka, the `#call` method is expect
     </tr>
   </tbody>
 </table>
+
+## Dynamic DLQ Target Topic
+
+Karafka Pro also supports the dynamic determination of the DLQ target topic. This feature is useful when the target DLQ topic may vary depending on runtime conditions or message metadata.
+
+To enable dynamic DLQ target topics, set the `topic:` option to `:strategy` in your routing configuration. Your strategy class's `#call` method should then return an array instead of a single symbol:
+
+- The first element is the symbol representing the action (`:retry`, `:dispatch`, `:skip`).
+- The second element specifies the dynamically determined target DLQ topic.
+
+```ruby
+class DynamicDlqStrategy
+  def call(errors_tracker, attempt)
+    if errors_tracker.last.is_a?(SpecialError)
+      [:dispatch, 'dlq_topic_for_specials']
+    else
+      [:dispatch, 'dlq_topic_for_anything_else']
+    end
+  end
+end
+
+class KarafkaApp < Karafka::App
+  routes.draw do
+    topic :orders_states do
+      consumer OrdersStatesConsumer
+
+      dead_letter_queue(
+        topic: :strategy,
+        strategy: DynamicDlqStrategy.new
+      )
+    end
+  end
+end
+```

Pro/Virtual-Partitions.md

@@ -82,12 +82,27 @@ Below is a list of arguments the `#virtual_partitions` topic method accepts.
       <td><code>#call</code></td>
       <td>Reducer for VPs key. It allows for a custom reducer to achieve enhanced parallelization when the default reducer is insufficient.</td>
     </tr>
+    <tr>
+      <td><code>distribution</code></td>
+      <td>Symbol</td>
+      <td>
+        Strategy used to distribute messages across virtual partitions:
+        <ul style="margin-top: 10px;">
+          <li>
+            <code>:consistent</code> (default) ensures messages with the same key always go to the same virtual partition, maintaining consistency across batches.
+          </li>
+          <li>
+            <code>:balanced</code> distributes work evenly across workers while preserving message order within key groups, improving utilization by up to 50% for uneven workloads.
+          </li>
+        </ul>
+      </td>
+    </tr>
   </tbody>
 </table>
 
 ## Messages Distribution
 
-Message distribution is based on the outcome of the `virtual_partitions` settings. Karafka will make sure to distribute work into jobs with a similar number of messages in them (as long as possible). It will also take into consideration the current `concurrency` setting and the `max_partitions` setting defined within the `virtual_partitions` method.
+Message distribution is based on the outcome of the `virtual_partitions` settings. Karafka will make sure to distribute work into jobs with a similar number of messages in them (as long as possible). It will also take into consideration the current `concurrency` setting and the `max_partitions` setting defined within the `virtual_partitions` method and will take into consideration appropriate `:strategy`.
 
 Below is a diagram illustrating an example partitioning flow of a single partition data. Each job will be picked by a separate worker and executed in parallel (or concurrently when IO is involved).
 
@@ -187,6 +202,127 @@ routes.draw do
 end
 ```
 
+### Distribution Strategies
+
+Karafka's Virtual Partitions feature provides two distribution strategies to determine how messages are allocated across consumer instances:
+
+- `:consistent` (default)
+- `:balanced`.
+
+These strategies give you flexibility in optimizing message distribution based on your specific workload characteristics and processing approach.
+
+#### Consistent Distribution (Default)
+
+By default, Karafka uses a consistent distribution strategy that ensures messages with the same partitioner result are always assigned to the same virtual partition consumer. This provides predictable and stable message routing, particularly important for stateful processing or when message order within a key group must be preserved across multiple batches.
+
+```ruby
+routes.draw do
+ topic :orders_states do
+   consumer OrdersStatesConsumer
+
+   virtual_partitions(
+     partitioner: ->(message) { message.headers['order_id'] },
+     # Default - each key always gets routed to the same virtual partition
+     # This provides consistent multi-batch distribution
+     distribution: :consistent
+   )
+ end
+end
+```
+
+The consistent distribution strategy ensures that:
+
+1. The same virtual partition always processes messages with the same partitioner outcome
+2. Distribution remains stable between batches
+3. Per-key ordering is strictly maintained
+
+However, consistent distribution can sometimes lead to suboptimal resource utilization when certain keys contain significantly more messages than others, potentially leaving some worker threads idle while others are overloaded.
+
+#### Balanced Distribution
+
+Karafka also supports a balanced distribution strategy that dynamically distributes workloads across available workers, potentially improving resource utilization by up to 50%. This strategy prioritizes even work distribution while maintaining message order within each key group.
+
+```ruby
+routes.draw do
+ topic :orders_states do
+   consumer OrdersStatesConsumer
+
+   virtual_partitions(
+     partitioner: ->(message) { message.headers['order_id'] },
+     # Balanced distribution for more even workload distribution
+     distribution: :balanced
+   )
+ end
+end
+```
+
+The balanced distribution strategy operates as follows:
+
+1. Messages are grouped by their partition key (as determined by the partitioner)
+2. Key groups are sorted by size (number of messages) in descending order
+3. Each key group is assigned to the worker with the least current workload
+4. Messages within each group maintain their offset order
+
+This approach ensures that:
+
+- Larger message groups are processed first
+- Work is distributed more evenly across available workers
+- Message order within each key group is preserved within a single batch
+- All available worker threads are utilized effectively
+
+##### Important Considerations for Balanced Distribution
+
+When using the balanced distribution strategy, keep in mind:
+
+- **Cross-batch assignment is not guaranteed** - Unlike consistent distribution, the same key may be assigned to different virtual partitions across different batches
+- **Stateful processing considerations** - If your consumer maintains state for specific keys across multiple batches, consistent distribution may still be more appropriate
+- **Messages with the same key are never split** - While keys may be assigned to different virtual partitions in different batches, all messages with the same key in a single batch will be processed together
+
+#### Choosing the Right Distribution Strategy
+
+Consider these factors when selecting a distribution strategy:
+
+<table border="1">
+  <thead>
+    <tr>
+      <th>Use <code>:consistent</code> when:</th>
+      <th>Use <code>:balanced</code> when:</th>
+    </tr>
+  </thead>
+  <tbody>
+    <tr>
+      <td>Processing requires stable assignment of keys to workers across batches</td>
+      <td>Processing is stateless or state is managed externally</td>
+    </tr>
+    <tr>
+      <td>You're implementing window-based aggregations spanning multiple polls</td>
+      <td>Maximizing worker thread utilization is a priority</td>
+    </tr>
+    <tr>
+      <td>Predictable routing is more important than even utilization</td>
+      <td>Message keys have highly variable message counts</td>
+    </tr>
+    <tr>
+      <td>Keys have relatively similar message counts</td>
+      <td>You want to optimize for throughput with uneven workloads</td>
+    </tr>
+  </tbody>
+</table>
+
+#### Performance Comparison
+
+The balanced distribution strategy can significantly improve resource utilization in high-throughput scenarios with uneven message distribution. Internal benchmarks show improvements of up to 50% in throughput for workloads where:
+
+- Message keys have highly variable message counts
+- Processing is IO-bound (such as database operations)
+- Worker threads would otherwise be underutilized with consistent distribution
+
+The performance gains are most significant when:
+
+1. Some keys contain many more messages than others
+2. The total number of keys is greater than the number of available worker threads
+3. Message processing involves IO operations that can benefit from concurrent execution
+
 ## Managing Number of Virtual Partitions
 
 By default, Karafka will create at most `Karafka::App.config.concurrency` concurrent Virtual Partitions. This approach allows Karafka to occupy all the threads under optimal conditions.

WaterDrop/Usage.md

@@ -51,6 +51,63 @@ Here are all the things you can provide in the message hash:
 
 Keep in mind, that message you want to send should be either binary or stringified (to_s, to_json, etc).
 
+## Headers
+
+Kafka headers allow you to attach key-value metadata to messages, which can be helpful for routing, filtering, tracing, and more. WaterDrop supports headers via the `headers:` key in message hashes.
+
+### Format
+
+Kafka headers are optional and must be provided as a `Hash`. According to [KIP-82](https://cwiki.apache.org/confluence/display/KAFKA/KIP-82+-+Add+Record+Headers), each header key must be a string, and each value must be either:
+
+- a **string**, or
+- an **array of strings**.
+
+This means WaterDrop supports both forms:
+
+```ruby
+# Single value per header
+headers: {
+  'request-id' => '123abc',
+  'source' => 'payment-service'
+}
+```
+
+```ruby
+# Multiple values per header key (KIP-82-compliant)
+headers: {
+  'flags' => ['internal', 'async'],
+  'source' => ['payment-service']
+}
+```
+
+### Example Usage
+
+#### Sync with headers
+
+```ruby
+producer.produce_sync(
+  topic: 'my-topic',
+  payload: 'payload-with-headers',
+  headers: {
+    'request-id' => 'abc-123',
+    'tags' => ['blue', 'fast']
+  }
+)
+```
+
+#### Async with headers
+
+```ruby
+producer.produce_async(
+  topic: 'my-topic',
+  payload: 'payload-with-headers',
+  headers: {
+    'tenant-id' => 'tenant-42',
+    'features' => ['beta', 'test']
+  }
+)
+```
+
 ## Delivery Results
 
 When dispatching messages using WaterDrop, you can choose between receiving a delivery report or a delivery handle, depending on whether you perform synchronous or asynchronous dispatches.