Kafka #

# ruby-kafka — the Zendesk gem, easier to use
gem install ruby-kafka

# rdkafka-ruby — C extension based on librdkafka, faster
# librdkafka must be installed first
sudo apt install librdkafka-dev   # Ubuntu
brew install librdkafka           # macOS
gem install rdkafka

# Karafka — a complete framework for Rails
gem install karafka

# Gemfile
gem 'ruby-kafka',  '~> 1.5'    # option 1: ruby-kafka
gem 'rdkafka',     '~> 0.14'   # option 2: rdkafka (higher performance)
gem 'karafka',     '~> 2.3'    # option 3: Karafka (framework)
gem 'avro_turf',   '~> 1.7'    # for Avro serialization (optional)
gem 'waterdrop',   '~> 2.6'    # standalone producer from the Karafka team

require 'kafka'
require 'json'

# Create a Kafka client
kafka = Kafka.new(
  ["localhost:9092"],            # broker list
  client_id:          "store-app",
  logger:             Logger.new($stdout),
  ssl_ca_cert:        ENV["KAFKA_CA_CERT"],  # optional, for SSL
  sasl_plain_username: ENV["KAFKA_USER"],    # optional, for authentication
  sasl_plain_password: ENV["KAFKA_PASSWORD"]
)

# Simple producer — a single message
kafka.deliver_message(
  JSON.generate({
    event:     "order_created",
    order_id:  12345,
    user_id:   99,
    total:     150_000,
    timestamp: Time.now.iso8601
  }),
  topic: "orders",
  key:   "order-12345"   # the key determines which partition is used
)

# Producer with batching — more efficient for high throughput
producer = kafka.producer(
  required_acks:      :all,        # wait for all replicas to acknowledge
  ack_timeout:        5,           # timeout in seconds
  max_buffer_size:    1000,        # max messages in the buffer
  max_buffer_bytesize: 1_000_000,  # max buffer size in bytes
  compression_codec:  :snappy      # compress with snappy (saves bandwidth)
)

begin
  # Add messages to the buffer
  100.times do |i|
    producer.produce(
      JSON.generate({ event: "event_#{i}", data: "payload" }),
      topic: "events",
      key:   "key-#{i % 10}",          # distribute to 10 partitions
      partition_key: "user-#{i % 5}"   # or use partition_key
    )
  end

  # Send all messages in the buffer to Kafka
  producer.deliver_messages
ensure
  producer.shutdown
end

require 'rdkafka'
require 'json'

config = Rdkafka::Config.new(
  "bootstrap.servers"        => "localhost:9092",
  "client.id"                => "store-app",
  "acks"                     => "all",
  "enable.idempotence"       => true,        # idempotent producer
  "compression.type"         => "snappy",
  "batch.size"               => 65_536,      # 64KB per batch
  "linger.ms"                => 5,           # wait 5ms before sending
  "message.max.bytes"        => 1_048_576    # max 1MB per message
)

producer = config.producer

# Send a message asynchronously
delivery_handle = producer.produce(
  topic:   "orders",
  key:     "order-12345",
  payload: JSON.generate({
    event:     "order_created",
    order_id:  12345,
    total:     150_000,
    timestamp: Time.now.to_i
  })
)

# Wait for confirmation (optional — can also be fire-and-forget)
delivery_report = delivery_handle.wait(max_wait_timeout: 10)
puts "Offset: #{delivery_report.offset}, Partition: #{delivery_report.partition}"

# Send many messages
handles = []
1000.times do |i|
  handles << producer.produce(
    topic:   "events",
    key:     "user-#{i % 100}",
    payload: JSON.generate({ seq: i, ts: Time.now.to_i })
  )
end

# Flush — wait until everything is sent
producer.flush(10_000)
producer.close

require 'kafka'
require 'json'

kafka = Kafka.new(["localhost:9092"], client_id: "notification-service")

# Consumer group — automatic load balancing between consumer instances
consumer = kafka.consumer(
  group_id:          "notification-service",
  offset_commit_interval: 5,    # commit offsets every 5 seconds
  offset_commit_threshold: 100, # or every 100 messages
  heartbeat_interval: 10,       # heartbeat to Kafka every 10 seconds
  session_timeout:    30        # session timeout if no heartbeat
)

# Subscribe to one or more topics
consumer.subscribe("orders", start_from_beginning: false)
consumer.subscribe("payments")

# Trap signals for graceful shutdown
trap("SIGTERM") { consumer.stop }
trap("SIGINT")  { consumer.stop }

# Message reading loop
consumer.each_message do |message|
  puts "Topic: #{message.topic}"
  puts "Partition: #{message.partition}"
  puts "Offset: #{message.offset}"
  puts "Key: #{message.key}"

  begin
    data = JSON.parse(message.value, symbolize_names: true)
    process_event(data)
  rescue JSON::ParserError => e
    puts "Invalid JSON: #{e.message}"
    # Send to a dead letter queue or log for analysis
  rescue => e
    puts "Error processing message: #{e.message}"
    # Don't raise — it would stop the consumer
    # Send to a DLQ if needed
  end
end

require 'rdkafka'
require 'json'

config = Rdkafka::Config.new(
  "bootstrap.servers"  => "localhost:9092",
  "group.id"           => "analytics-service",
  "auto.offset.reset"  => "earliest",      # start from the beginning if the group is new
  "enable.auto.commit" => false,           # manual commit for full control
  "max.poll.interval.ms" => 300_000,       # max 5 minutes per poll
  "session.timeout.ms"   => 30_000         # 30 second session timeout
)

consumer = config.consumer
consumer.subscribe("orders")

# Graceful shutdown
running = true
trap("SIGTERM") { running = false }
trap("SIGINT")  { running = false }

begin
  while running
    # Poll with a 1 second timeout
    message = consumer.poll(1000)
    next unless message

    begin
      data = JSON.parse(message.payload, symbolize_names: true)
      process_event(data)

      # Commit the offset AFTER successful processing
      # This ensures messages aren't lost if a crash occurs before processing
      consumer.commit
    rescue => e
      puts "Error: #{e.message}"
      # Don't commit — the message will be re-read when the consumer restarts
      send_to_dlq(message) if attempts_for?(message) >= 3
    end
  end
ensure
  consumer.close
end

require 'kafka'
require 'json'

class KafkaConsumerWithDLQ
  MAX_RETRY = 3
  DLQ_TOPIC = "orders.dlq"

  def initialize(kafka)
    @kafka    = kafka
    @consumer = kafka.consumer(group_id: "order-service")
    @producer = kafka.producer
    @retry_counts = {}
  end

  def run
    @consumer.subscribe("orders")

    @consumer.each_message do |message|
      process_with_retry(message)
    end
  ensure
    @producer.shutdown
  end

  private

  def process_with_retry(message)
    key = "#{message.topic}-#{message.partition}-#{message.offset}"
    @retry_counts[key] ||= 0

    begin
      data = JSON.parse(message.value, symbolize_names: true)
      process_order(data)
      @retry_counts.delete(key)

    rescue => e
      @retry_counts[key] += 1

      if @retry_counts[key] >= MAX_RETRY
        send_to_dlq(message, e)
        @retry_counts.delete(key)
      else
        puts "Attempt #{@retry_counts[key]}/#{MAX_RETRY}: #{e.message}"
        sleep(2 ** @retry_counts[key])  # exponential backoff: 2s, 4s, 8s
        retry
      end
    end
  end

  def send_to_dlq(original_message, error)
    @producer.produce(
      JSON.generate({
        original_topic:     original_message.topic,
        original_partition: original_message.partition,
        original_offset:    original_message.offset,
        original_key:       original_message.key,
        original_payload:   original_message.value,
        error_class:        error.class.name,
        error_message:      error.message,
        failed_at:          Time.now.iso8601
      }),
      topic: DLQ_TOPIC,
      key:   original_message.key
    )
    @producer.deliver_messages
    puts "Message sent to DLQ: #{DLQ_TOPIC}"
  end
end

# Initialize Karafka in a Rails project
bundle exec karafka install

# karafka.rb — main configuration
class KarafkaApp < Karafka::App
  setup do |config|
    config.kafka = {
      "bootstrap.servers"       => ENV.fetch("KAFKA_BROKERS", "localhost:9092"),
      "client.id"               => "store-app",
      "group.id"                => "store-consumers",
      "enable.idempotence"      => true,
      "compression.type"        => "snappy",
      "session.timeout.ms"      => 30_000,
      "max.poll.interval.ms"    => 300_000,
      "enable.auto.commit"      => false   # Karafka manages commits itself
    }

    config.consumer_persistence = !Rails.env.test?
    config.logger = Rails.logger
  end

  # Routing — defines a consumer for every topic
  routes.draw do
    topic :orders do
      consumer OrdersConsumer
    end

    topic :payments do
      consumer PaymentsConsumer
      max_messages  100          # process max 100 messages per batch
      max_wait_time 5_000        # wait max 5 seconds for a batch
    end

    topic :notifications do
      consumer NotificationsConsumer
      manual_offset_management true  # manual commit
    end

    # A different consumer group for the same topic
    consumer_group :analytics do
      topic :orders do
        consumer OrdersAnalyticsConsumer
      end
    end
  end
end

# app/consumers/orders_consumer.rb
class OrdersConsumer < ApplicationConsumer
  # messages — array of Karafka::Messages::Message
  def consume
    messages.each do |message|
      data = message.payload   # Karafka automatically parses JSON

      Rails.logger.info "Processing order: #{data['order_id']}"

      Order.transaction do
        order = Order.find_or_create_by!(id: data['order_id'])
        order.update!(
          status:     data['status'],
          total:      data['total'],
          processed_at: Time.now
        )

        # Send a notification
        NotificationMailer.order_confirmed(order).deliver_later
      end
    end

    # Karafka automatically commits after a successful consume
  end

  # Called when an exception occurs in consume
  def handle_exception(exception)
    Rails.logger.error "Order consumer error: #{exception.message}"
    # Karafka will retry automatically according to the configuration
  end
end

# app/consumers/payments_consumer.rb
class PaymentsConsumer < ApplicationConsumer
  def consume
    # Batch processing — all messages processed at once
    payment_ids = messages.payloads.map { |p| p['payment_id'] }
    Rails.logger.info "Processing #{messages.count} payments: #{payment_ids}"

    Payment.where(id: payment_ids).find_each do |payment|
      payment.confirm!
    end
  end
end

# config/initializers/waterdrop.rb
WaterDrop.setup do |config|
  config.logger = Rails.logger
  config.deliver = !Rails.env.test?

  config.kafka = {
    "bootstrap.servers":  ENV.fetch("KAFKA_BROKERS", "localhost:9092"),
    "acks":               "all",
    "enable.idempotence": true,
    "compression.type":   "snappy"
  }
end

# Send messages from anywhere in Rails
class OrderService
  def create_order(user, item)
    order = Order.create!(user: user, item: item)

    # Send an event to Kafka
    WaterDrop::Producer.new.call do |producer|
      producer.produce_sync(
        topic:   "orders",
        key:     order.id.to_s,
        payload: {
          event:     "order_created",
          order_id:  order.id,
          user_id:   user.id,
          total:     order.total,
          timestamp: Time.now.iso8601
        }.to_json
      )
    end

    order
  end
end

# Or use a producer created once (more efficient)
KAFKA_PRODUCER = WaterDrop::Producer.new
KAFKA_PRODUCER.setup do |config|
  config.kafka = { "bootstrap.servers": "localhost:9092" }
end

# Singleton producer — reuse across the whole application
KAFKA_PRODUCER.produce_async(
  topic:   "events",
  payload: { event: "user_login", user_id: 1 }.to_json
)

# Batch produce — more efficient
messages = 1000.times.map do |i|
  { topic: "events", payload: { seq: i }.to_json }
end
KAFKA_PRODUCER.produce_many_async(messages)

require 'avro_turf'
require 'avro_turf/messaging'

# Schema Registry — a server storing Avro schemas
avro = AvroTurf::Messaging.new(
  registry_url: ENV.fetch("SCHEMA_REGISTRY_URL", "http://localhost:8081")
)

# Avro schema (usually defined in the Schema Registry)
# {
#   "type": "record",
#   "name": "Order",
#   "fields": [
#     {"name": "order_id", "type": "int"},
#     {"name": "total", "type": "double"},
#     {"name": "status", "type": "string"}
#   ]
# }

# Encode a message with Avro
avro_payload = avro.encode(
  { "order_id" => 12345, "total" => 150_000.0, "status" => "pending" },
  subject: "orders-value",
  version: :latest
)

# Send the encoded payload
producer.produce(topic: "orders", payload: avro_payload, key: "12345")

# Decode on the consumer side
message = consumer.poll(1000)
data = avro.decode(message.payload)
puts data["order_id"]   # => 12345

# Idempotent producer — Kafka guarantees no duplicates even if the producer retries
config = Rdkafka::Config.new(
  "bootstrap.servers" => "localhost:9092",
  "enable.idempotence" => true,   # enable idempotence
  # enable.idempotence automatically sets:
  # acks = all
  # max.in.flight.requests.per.connection = 5
  # retries = INT_MAX
)

# Exactly-once semantics — read from Kafka, process, write to Kafka
# exactly once (no duplicates, no losses)
config_eos = Rdkafka::Config.new(
  "bootstrap.servers"            => "localhost:9092",
  "enable.idempotence"           => true,
  "transactional.id"             => "store-processor-1",  # unique ID per instance
  "transaction.timeout.ms"       => 60_000
)

producer_eos = config_eos.producer
producer_eos.init_transactions

input_consumer = config.consumer
input_consumer.subscribe("raw_orders")

loop do
  message = input_consumer.poll(1000)
  next unless message

  begin
    producer_eos.begin_transaction

    # Process and transform the message
    processed_data = transform(JSON.parse(message.payload))

    # Write the result to another topic within the same transaction
    producer_eos.produce(
      topic:   "processed_orders",
      payload: processed_data.to_json,
      key:     message.key
    )

    # Commit the consumer offset as part of the transaction
    producer_eos.send_offsets_to_transaction(
      input_consumer,
      "processor-group"
    )

    producer_eos.commit_transaction

  rescue => e
    producer_eos.abort_transaction
    puts "Transaction aborted: #{e.message}"
  end
end

# Statistics from rdkafka
config = Rdkafka::Config.new(
  "bootstrap.servers"  => "localhost:9092",
  "statistics.interval.ms" => 5_000  # report every 5 seconds
)

config.statistics_callback = proc do |stats|
  stats_hash = JSON.parse(stats)

  # Log consumer lag — how far behind the consumer is
  stats_hash["topics"]&.each do |topic, topic_data|
    topic_data["partitions"]&.each do |partition, part_data|
      lag = part_data["consumer_lag"]
      next if lag < 0   # -1 means unknown

      puts "Lag #{topic}[#{partition}]: #{lag} messages"

      # Alert if the lag is too large
      if lag > 10_000
        Monitoring.alert("High Kafka consumer lag: #{topic}[#{partition}] = #{lag}")
      end
    end
  end
end

# Important metrics to monitor:
# consumer_lag      — number of unprocessed messages
# messages_per_sec  — producer throughput
# bytes_per_sec     — bandwidth
# error_count       — number of errors