Kafka Client Benchmarks

This repository contains performance and stress benchmarks for various Kafka client implementations across Crystal (Kafkaesque) and Go (Confluent Go, Franz-Go).

---

Stress Benchmark Configuration

• Message Count: 1,000,000 messages
• Linger MS: 100 ms
• Outlier Mitigation: Running 3 iterations of each test and selecting the median run.
• Warmup: A silent warmup phase of 100,000 messages to prime client runtimes and OS socket buffers.
• Coexistence / Zero Redundant Build: Single Crystal binaries are compiled once (-Dpreview_mt enabled) and runtime worker threads are controlled dynamically via the CRYSTAL_WORKERS environment variable to avoid redundant compilation phases.
• Environment Execution Modes:
  1. Single-Core (Stage 1): Pinned to a single core (taskset -c 2) with CRYSTAL_WORKERS=1 for single-thread targets.
  2. Multi-Core (Stage 2): Scaled across all 8 cores (taskset -c 0-7) with CRYSTAL_WORKERS=8 for multi-threaded targets.

---

Benchmark Results

1. Single-Core Results (Pinned to Core 2)

Producer Scoreboard

Consumer Scoreboard

---

2. Multi-Core Results (Pinned to Cores 0-7)

Producer Scoreboard

Consumer Scoreboard

---

Client Configurations

Below are the exact code configuration settings used for each client engine during the benchmarks:

1. Kafkaesque (Crystal)

Example Console Startup Output:

🚀 Kafkaesque Producer configured and ready (PLAINTEXT, port 9097)
--- Verbose Configurations ---
Bootstrap Servers: localhost:9097
Settings:
  enable.idempotence: true
  acks: all
  linger.ms: 100
  batch.num.messages: 10000
  retries: 5
  retry.backoff.ms: 100
  compression.type: lz4
------------------------------

🚀 Kafkaesque Consumer started (PLAINTEXT, port 9097)...
--- Verbose Configurations ---
Bootstrap Servers: localhost:9097
Settings:
  group.id: bench-kafkaesque-178047...
  initial_offset_smallest: true
  group.protocol: consumer
------------------------------

Producer Setup

producer_config = Kafkaesque::Producer::Config.new(
  bootstrap_servers: ["localhost:9097"],
  compression_type: "lz4",
  settings: {
    "enable.idempotence" => "true",
    "acks"               => "all",
    "linger.ms"          => "100",
    "batch.num.messages" => "10000",
    "retries"            => "5",
    "retry.backoff.ms"   => "100",
  }
)

Consumer Setup

config = Kafkaesque::Consumer::Config.new(
  bootstrap_servers: ["localhost:9097"],
  group_id: "bench-kafkaesque-[timestamp]",
  initial_offset_smallest: true
)
# Note: Defaults to auto-commit enabled and uses "group.protocol": "consumer" (KIP-848 protocol)

2. Go Confluent (Go)

Example Console Startup Output:

🚀 Go Confluent Producer configured and ready (PLAINTEXT, port 9097)
   Settings: enable.idempotence=true, acks=all, linger.ms=100, batch.num.messages=10000, compression=lz4, go.delivery.reports=false

🚀 Go Confluent Consumer started (PLAINTEXT, port 9097)...
   Settings: group.id=bench-go-confluent-178047..., auto.offset.reset=smallest, enable.auto.commit=true, group.protocol=consumer, fetch.wait.max.ms=5

Producer Setup

p, err := kafka.NewProducer(&kafka.ConfigMap{
    "bootstrap.servers":   "localhost:9097",
    "compression.type":    "lz4",
    "enable.idempotence":  true,
    "acks":                "all",
    "linger.ms":           100,
    "batch.num.messages":  10000,
    "retries":             5,
    "retry.backoff.ms":    100,
    "go.delivery.reports": false,
})

Consumer Setup

c, err := kafka.NewConsumer(&kafka.ConfigMap{
    "bootstrap.servers":  "localhost:9097",
    "group.id":           "bench-go-confluent-[timestamp]",
    "auto.offset.reset":  "smallest",
    "enable.auto.commit": true,
    "group.protocol":     "consumer",
    "fetch.wait.max.ms":  5,
})

3. Franz-Go (Go)

Example Console Startup Output:

🚀 Franz-Go Producer configured and ready (PLAINTEXT, port 9097)
   Settings: RequiredAcks=all, linger.ms=100, batch.num.messages=10000 (1MB limit), compression=lz4

🚀 Franz-Go Consumer started (PLAINTEXT, port 9097)...
   Settings: group.id=bench-go-franz-178047..., auto.offset.reset=smallest (default), group.protocol=consumer

Producer Setup

opts := []kgo.Opt{
    kgo.SeedBrokers("localhost:9097"),
    kgo.ProducerLinger(time.Duration(lingerMs) * time.Millisecond),
    kgo.ProducerBatchMaxBytes(1000000),
    kgo.RequiredAcks(kgo.AllISRAcks()),
    kgo.ProducerBatchCompression(kgo.Lz4Compression()),
}

Consumer Setup

opts := []kgo.Opt{
    kgo.SeedBrokers("localhost:9097"),
    kgo.ConsumerGroup("bench-go-franz-[timestamp]"),
    kgo.ConsumeTopics("test-topic"),
    kgo.GroupProtocol("consumer"),
}