Cascade

I'm building a fast Kafka-style streaming log in pure Scala 3.

I built Cascade around the Kafka wire protocol so existing Kafka clients can connect without a custom SDK. If a language has a client that speaks one of the supported protocol versions, it can talk to Cascade.

The broker itself only needs Scala and the JDK. I use Apache Kafka's Java client in the test suite as an independent compatibility check; it isn't a runtime dependency.

So far, I've implemented broker-assigned offsets, magic-v2 record batches, consumer coordination, durable metadata and offset journals, idempotent producer recovery, transactions, read_committed isolation, ISR replication, partition-leader promotion, and quorum controller election.

I measured this on my Windows development machine with Java 21, eight partitions, four producers, four consumers, 1 KiB deterministic incompressible payloads, LZ4, acks=all, and periodic background flushing:

After I fixed the background-flush path, sustained ten-million-record production went from 57,400 to 182,285 records/s: a 3.18x improvement. The write phase dropped from 174.2 to 54.9 seconds. That run forced 9.58 GiB in 47.6 cumulative seconds, so the drive was the main limit on this machine.

The one-million test is much shorter and benefits a lot from the filesystem cache. I don't present either result as production capacity. The full heavy-load report includes the machine, workload, latency, CPU, GC, heap, storage, and test method.

• Persistent, length-delimited, big-endian Kafka TCP frames.
• Hard request-size bounds and version validation before request handling.
• Correlation IDs preserved in every response.
• Ordered processing within a connection and Java 21 virtual-thread isolation between connections.
• Explicitly advertised API keys and versions rather than a broad, unverified compatibility claim.
• Kafka 4.3.1 Admin, Producer, transactional Producer, explicit Consumer, and subscribed Consumer interoperability tests.

Any language can connect if its client speaks one of the supported Kafka protocol versions. For now, my automated end-to-end tests use the Kafka Java client. I still need to add proper client matrices for Python, Go, .NET, and other languages.

• Immutable Kafka magic-v2 record batches stored without broker-side decompression or recompression.
• Client compression remains compatible with Snappy, LZ4, Zstd, and gzip because payloads stay opaque to storage.
• Serialized broker offset assignment per partition.
• Batch-aligned Fetch responses and correct first-batch size behavior.
• Size-based log segments and startup index reconstruction by scanning batch headers.
• Incomplete-tail truncation and removal of later segments whose offsets depend on a damaged earlier tail.
• periodic background flushing with dirty-age, byte-threshold, and segment-rollover triggers.
• sync flushing for strict per-append local persistence.
• Clean shutdown forces all remaining dirty segments.

Kafka acknowledgements and local disk forcing are separate settings. In single-node periodic mode, acks=1 and acks=all acknowledge a local append before the next scheduled force. A process, OS, or power failure before that force can lose those records. In cluster mode, acks=all requires the configured minimum ISR and waits for every current ISR member to append before the committed high watermark advances.

In single-node mode I support this Kafka delivery-semantics path:

• Durable producer ID allocation and producer-epoch fencing.
• Per-partition sequence validation with wraparound after Int.MaxValue.
• A bounded five-batch duplicate window that returns the original offset for a retry.
• O(1) lookup of recent producer batches on the append hot path.
• Producer sequence recovery from persisted record-batch headers after restart.
• Durable active transaction, outcome, range, timeout, and transactional-offset state.
• Transaction commit and abort, including safe coordination with in-flight appends.
• Last stable offsets and read_committed filtering of aborted or open transactions.
• Transaction timeouts and automatic abort when a new epoch fences an old owner.
• Applied checkpoints that recover an interrupted transactional offset commit without replaying old transactions over newer offsets.

This gives Cascade the building blocks for exactly-once processing on one broker. I haven't replicated producer and transaction coordinator state in static-cluster mode, so I'm not claiming cross-broker delivery failover yet.

• Classic Kafka group protocol with member ID allocation.
• Join and sync phases, protocol selection, and leader-provided assignments.
• Heartbeats, session expiration, explicit leave, and rebalance initiation.
• Generation/member validation for ordinary offset commits.
• Append-only CRC32C-protected offset journal with one forced commit per multi-partition request.
• Offset recovery across restart and partial/corrupt-tail truncation.
• Transactional offset staging and commit through TxnOffsetCommit.

If a client exposes Kafka's newer consumer protocol, set group.protocol=classic for now.

• Majority-committed, monotonic metadata images.
• Durable controller terms and one vote per term in a forced CRC32C journal.
• Majority election with metadata-freshness voting, randomized retry deadlines, and a preferred initial candidate.
• Controller heartbeats and leases that fence isolated or not-yet-synchronized brokers.
• A committed controller-term image that bumps every partition leader epoch after election.
• Dynamic controller discovery in Kafka Metadata and coordinator responses.
• Round-robin replica assignment and configurable default replication factor.
• Partition leaders, leader epochs, replica sets, and ISR state.
• Parallel synchronous leader-to-follower append.
• min.insync.replicas admission for acks=all.
• Leader-only Fetch/ListOffsets and committed high-watermark visibility.
• Failure detection, ISR shrink, and promotion of a surviving replica with a new leader epoch.
• Full committed-prefix recovery for a returning replica, including replacement of a divergent local tail.
• Partition fencing during recovery so Produce cannot race the copy or ISR admission.
• ISR re-admission only after catch-up succeeds and the new metadata image reaches quorum.
• Real Kafka-client end-to-end verification across partition-leader loss, controller loss, stale-term rejection, metadata creation after election, and broker restart/rejoin.

Broker membership is still static, but the controller is elected from that configured quorum. Recovery currently copies the complete committed log instead of using an incremental snapshot or segment transfer, so it favors correctness over recovery bandwidth. Group, offset, producer, and transaction coordinator state is still local to one broker, so those services don't have production failover yet.

• JDK 21+
• Network access to Maven Central for the first build

The included launchers download SBT the first time you run them.

Windows:

.\sbt.bat test
.\sbt.bat "run --host 0.0.0.0 --port 9092 --advertised-host localhost --data-dir data"

Linux/macOS:

chmod +x sbt
./sbt test
./sbt "run --host 0.0.0.0 --port 9092 --advertised-host localhost --data-dir data"

Point any supported Kafka client at localhost:9092. A basic single-node producer configuration is:

bootstrap.servers=localhost:9092
acks=all
enable.idempotence=true

For a consumer using subscriptions and committed-only visibility:

bootstrap.servers=localhost:9092
group.protocol=classic
isolation.level=read_committed

Set transactional.id on a producer when you want to use transactions. Use read_uncommitted only when the consumer should also see aborted transactional records.

Run each command in a separate process and give every broker its own data directory:

.\sbt.bat "run --host 127.0.0.1 --port 9092 --advertised-host 127.0.0.1 --advertised-port 9092 --node-id 1 --data-dir data-1 --cluster-nodes 1@127.0.0.1:9092,2@127.0.0.1:9093,3@127.0.0.1:9094 --controller-id 1 --default-replication-factor 3 --min-insync-replicas 2"
.\sbt.bat "run --host 127.0.0.1 --port 9093 --advertised-host 127.0.0.1 --advertised-port 9093 --node-id 2 --data-dir data-2 --cluster-nodes 1@127.0.0.1:9092,2@127.0.0.1:9093,3@127.0.0.1:9094 --controller-id 1 --default-replication-factor 3 --min-insync-replicas 2"
.\sbt.bat "run --host 127.0.0.1 --port 9094 --advertised-host 127.0.0.1 --advertised-port 9094 --node-id 3 --data-dir data-3 --cluster-nodes 1@127.0.0.1:9092,2@127.0.0.1:9093,3@127.0.0.1:9094 --controller-id 1 --default-replication-factor 3 --min-insync-replicas 2"

For now, keep enable.idempotence=false in static-cluster tests. I haven't replicated producer and transaction coordinator state yet.

Cascade returns exactly this matrix from ApiVersions:

I follow the Apache Kafka 4.3 protocol grammar. Cascade only advertises and accepts the versions listed above; I don't want to claim support for versions I haven't tested.

Kafka client in any language
        |
        | Kafka length-delimited TCP frames
        v
Virtual-thread connection handler
        |
        | frame bounds + versioned binary codec
        v
Request routing
        |
        +-- metadata quorum + durable controller election and fencing
        +-- classic group coordinator + durable offset journal
        +-- producer/transaction coordinator + delivery journal
        +-- partition log
              +-- opaque Kafka magic-v2 batches
              +-- broker-assigned offsets
              +-- bounded producer-state index
              +-- segment rollover and crash-tail recovery
              +-- background or synchronous flushing
              +-- ISR follower replication and committed watermark

I keep records serialized on the data path. Positional file I/O avoids shared channel-position races, each partition serializes offset assignment, and Fetch returns complete record batches. A transactional append reserves its range before storage. That stops EndTxn from racing an in-flight append and keeps the last stable offset behind open work.

records             10,000,000
payload             1,024 bytes, deterministic incompressible variants
partitions          8
producer clients    4
consumer clients    4
compression         LZ4
acks                all
batch.size          128 KiB
linger.ms           5
segment size        256 MiB
flush policy        periodic
flush interval      1,000 ms
flush byte limit    64 MiB per partition

I start the acknowledgement timer when each asynchronous send is offered, so saturation latency includes producer-side queueing. The producers offered data faster than my local temporary drive could force it during a sustained ten-gigabyte run. That's why p95 stayed above five seconds after the flush fix.

After adding the cluster path, I ran this test again. It reached 615,592 produced and 527,001 consumed records/s, with p99 acknowledgement latency at or below 500 ms and a 422.632 ms maximum. I didn't find a meaningful single-node regression, but this short cache-assisted test isn't a sustained storage benchmark.

./sbt "Test/runMain cascade.performance.LoadTest --records 10000000 --payload-bytes 1024 --partitions 8 --producers 4 --consumers 4 --compression lz4 --flush-policy periodic --flush-interval-ms 1000 --flush-bytes 67108864"

On Windows, replace ./sbt with .\sbt.bat. Add --keep-data if you want to keep the generated segment directory. I leave this harness out of normal sbt test runs because it's large and its result depends on the machine.

The harness reports:

• Produce and consume records/s and payload MiB/s.
• Acknowledgement p50, p95, p99, p99.9, and maximum latency.
• Process CPU, machine CPU, GC collections, and GC time.
• Stored bytes, bytes per record, force count/volume/time, and pending dirty bytes.
• Peak heap and exact consumed-record verification.

The complete 2026-08-05 report compares this result with the old per-request-force implementation and explains what changed.

The current sbt test suite passes 46/46 tests in three layers:

• Unit tests for binary codecs, bounds failures, record-batch metadata and sequence wrap, storage pagination, segment rollover, flush policies, corrupt/partial-tail recovery, replica reset, delivery-state recovery, producer fencing, transaction timeout, interrupted offset application, group coordination, metadata recovery, and durable controller term/vote recovery.
• TCP integration tests for discovery, Produce/Fetch, acknowledgement behavior, duplicate retry offsets, sequence-gap rejection, and idempotent state recovery after broker restart.
• Kafka 4.3.1 end-to-end tests for Admin/Producer/Consumer interoperability, classic group rebalances, committed offsets across restart, RF=3 replication, ISR/leader failover, controller election and stale-term fencing, divergent-tail replacement, safe replica re-admission, transactions, commit/abort isolation, active last stable offsets, and transactional consumer offsets.

The load harness separately checks the exact record count at one million and ten million records.

I track the release gates in docs/production-readiness.md. I won't call Cascade a production Kafka replacement until every blocking gate passes on the deployment topology I document.

• Broker and voter membership is statically configured; online membership changes and joint-consensus reconfiguration are not implemented.
• Replica recovery currently recopies the full committed prefix and pauses Produce for that partition; incremental segment transfer is not implemented yet.
• Group, offset, producer, and transaction coordinator state is not replicated for failover.
• Retention, compaction, timestamp indexes, quotas, TLS/SASL, ACLs, and operational endpoints are not implemented.
• Only the classic consumer group protocol is supported.
• The automated client matrix currently uses Kafka Java client 4.3.1.
• The performance figures are single-node, shared-JVM development-machine measurements; replicated-cluster capacity has not been benchmarked.

• Production-readiness gates
• Heavy-load report
• Contributing
• Apache-2.0 license

Apache License 2.0.