rb:karafka

Karafka

Apache Kafka for Ruby with high-performance multi-threaded processing.

Iron Laws

1. Process in batches when possible - Better throughput
2. Handle dead letter queues - Don't lose failed messages
3. Monitor lag and throughput - Key operational metrics
4. Use idempotent consumers - Safe to reprocess
5. Commit offsets after processing - At-least-once delivery
6. Partition by business key - Preserve ordering

Installation

gem 'karafka', '~> 2.5'
gem 'karafka-web'  # For Web UI

Note: Karafka 2.5+ is the current stable release. Check karafka.io for the latest version.

Basic Setup

# karafka.rb
class KarafkaApp < Karafka::App
  setup do |config|
    config.kafka = {
      'bootstrap.servers': ENV['KAFKA_BROKERS'],
      'client.id': 'my_app'
    }
    config.consumer_persistence = true
  end
  
  routes.draw do
    topic :orders do
      consumer OrdersConsumer
    end
    
    topic :payments do
      consumer PaymentsConsumer
    end
  end
end

KarafkaApp.boot!

Consumer

class OrdersConsumer < ApplicationConsumer
  def consume
    # Batch processing (recommended)
    Order.insert_all(messages.payloads)
    mark_as_consumed(messages.last)
    
    # Or individual processing
    messages.each do |message|
      process_payment(message.payload)
      mark_as_consumed(message)
    end
  end
end

class ApplicationConsumer < Karafka::BaseConsumer
  private
  
  def mark_as_consumed(message)
    consumer.mark_as_consumed(message)
  end
end

Producing Messages

# Sync (for critical messages)
Karafka.producer.produce_sync(
  topic: 'orders',
  payload: order.to_json,
  key: order.id  # Partition key
)

# Async (for high throughput)
Karafka.producer.produce_async(
  topic: 'orders',
  payload: order.to_json,
  key: order.id
)

# Batch
messages = orders.map { |o| { topic: 'orders', payload: o.to_json, key: o.id } }
Karafka.producer.produce_many_async(messages)