Point your Kafka client at Felix

Every Felix broker can speak Kafka. Set FELIX_KAFKA_LISTEN and the broker opens a second listener. Kafka producers write to durable Felix streams through it, and Kafka consumers read them back.

Producers can be idempotent, compress with any of Kafka's codecs, and pick acks 0, 1 or all. Consumers follow a partition to its new leader after a failover. It also stops somewhere specific. There are no consumer groups and no transactions, so a client built on group.id and subscribe() won't work.

How a Kafka request becomes a Felix operation

Kafka's model turned out to be close to Felix's already. A Kafka topic has partitions, each partition has one leader, and clients talk to the leader directly. A Felix stream has shards, each shard has one leader, and Felix clients talk to the leader directly. So the mapping is mostly naming.

A topic is <namespace>.<stream>, split on the first dot. Stream created in namespace orders is topic orders.created. The tenant never appears in the name. It comes from the SASL username, and the password is the same Felix token a native client sends.

Partition N is shard N. Offsets are Felix's own log offsets, with no translation. They start at 0 and are contiguous per shard, which is exactly the shape Kafka expects. That meant there was no offset table to keep durable, and no way for the two numbering schemes to drift apart.

Here is what each request turns into:

The listener lives in every broker rather than in a gateway in front, so Metadata just reports where each shard lives and the broker holding the log serves the fetch. A gateway would proxy every fetch and be one more thing to run.

Reads

A fetch is an offset-addressed range read, and the shard log already had one. When a fetch finds nothing new, it waits up to max_wait_ms (capped at 30 seconds), and the shard's append notification wakes it the moment a publish is durable.

When a shard moves or its leader fails, the old broker answers the next fetch with NOT_LEADER_OR_FOLLOWER. librdkafka refreshes its metadata, finds the new leader, and asks for the same offset there. The new leader holds the same log with the same offsets, so nothing is skipped or repeated. The consumer sees a short pause.

Writes

A produce goes through the same publish path a native Felix client uses, on the shard's leader. The broker decodes the record batch (gzip, snappy, lz4 and zstd all work) and caps decompression at 16 MiB per batch. Then it admits the write the way it admits a QUIC publish: it must hold its lease, routing must send the shard here, and the shard's write fence must be open. If any of that fails, the answer is NOT_LEADER_OR_FOLLOWER, and the client goes to the leader itself. Nothing is forwarded.

acks maps onto the stream's own consistency. With acks=0 there is no answer. With acks=1 the answer comes after the leader's write. With acks=all the answer waits for whatever the stream promises: a majority of replicas on a Quorum stream, or the leader alone on a Leader stream. Asking Kafka-style for “all” doesn't make a Leader stream into something else.

Keys, headers and producer timestamps are dropped, because a Felix record is a payload and an append time. The key has already chosen the partition by then, so same-key records still land in order on one shard. Headers are dropped rather than refused because tracing interceptors add them without asking.

Quotas are shared. A Kafka produce draws on the same per-tenant publish budget as the tenant's QUIC publishes (FELIX_TENANT_PUBLISH_BYTES_PER_SEC and friends). Over budget, the produce is still written and answered with throttle_time_ms, and the connection reads nothing more until that time passes. That matches what Kafka's own brokers do: they compute a delay, return it, and mute the channel, and the client holds off for the same window (Kafka docs, Quotas: Enforcement).

The hard part: a re-sent batch after a failover

Idempotence is on by default in Kafka's Java producer, and requires acks=all (Kafka producer configs, enable.idempotence). The promise is simple. If a producer sends a batch, loses the answer, and sends it again, the broker writes it once.

The hard case is when the answer is lost because the leader died. The producer re-sends to whichever broker took over. That broker has to know the batch is already there, with nothing from the dead leader to go on.

Felix already had idempotent producers of its own, and their state was in the right place. Each record carries its producer id and sequence in the shard log, replication ships those marks with the records, and a promoted replica rebuilds every producer's position from them. A batch re-sent to the new leader is answered with the offset it was first written at.

The catch was the numbering. Felix producers number batches: the log expects the next batch at the last sequence plus one. Kafka producers number records. KIP-98 says the sequence is “incremented by the producer on every message”, a batch carries the sequence of its first record, and the broker rejects anything not “exactly one greater than the last committed message” (KIP-98). So a Kafka batch of five records at base sequence 40 is followed by one at 45. Felix's log would call that a gap.

I could have taught the log a second counting rule. Instead the listener stores each record of an idempotent Kafka batch as a one-record Felix producer batch. Record i of a batch with base sequence s carries sequence s + i. The log's “one more than the last” is then exactly Kafka's rule, and everything the log does for Felix producers applies without changes (#710).

That choice also handled a case I hadn't planned for. If a leader died after shipping only the first three records of that batch, the new leader holds 40 to 42. The re-send asks for 40 to 44, the log says the producer owes 43, and only the last two are written. The answer carries the offset of record 40.

Kafka sequences are 31 bits and wrap to zero after i32::MAX; KIP-98 says they “wrap around once ... int32_max [is] reached”. Felix's log counts in 64 bits and never wraps. So the listener lifts each Kafka sequence to the 64-bit count nearest the one the producer owes:

pub(crate) fn lift(sequence: u64, owed: Option<u64>) -> u64 {
    let sequence = sequence % RECORD_SEQUENCE_WRAP;
    let Some(owed) = owed else {
        return sequence;
    };
    let base = owed - owed % RECORD_SEQUENCE_WRAP;
    [
        base.checked_sub(RECORD_SEQUENCE_WRAP),
        Some(base),
        base.checked_add(RECORD_SEQUENCE_WRAP),
    ]
    .into_iter()
    .flatten()
    .map(|base| base + sequence)
    .min_by_key(|candidate| candidate.abs_diff(owed))
    .expect("the base itself is always a candidate")
}

A re-send from either side of the wrap is then a duplicate, and a jump ahead is a gap, the same as anywhere else. Producing 2³¹ records in a test wasn't practical, so the wrap tests seed a producer near i32::MAX through the replication apply path, the state a promoted replica would hold. With lift reverted to return the raw sequence, both failed: the wrapped batch read as an old re-send (#784).

The failover test runs a three-broker Quorum stream. An idempotent producer writes with acks=all, the leader is killed, and the batch is re-sent to the promoted leader. Then the shard is moved and the batch is re-sent to the destination. A reader sees every record exactly once. With the log's duplicate answer switched off, the promoted leader appended the re-send at offset 6 instead of answering with its original offset 5, and the test failed.

Where it stops

There are no consumer groups. Kafka's group protocol means a coordinator, joins, syncs, heartbeats, committed offsets and rebalances. Felix has its own consumer groups for sharing work, used through a Felix client, but not Kafka's coordination protocol. That rules out more than subscribe(). Kafka Streams uses its application.id as “the group-id for membership management” (Streams configs), and Connect workers join a “Connect cluster group” through group.id (Connect configs). So Streams, Connect and the tools built on them won't run against Felix. A consumer that assigns its own partitions and keeps its own offsets works fine.

There are no transactions. A producer with transactional.id is refused, and so is anything exactly-once built on it.

There are no admin APIs. CreateTopics, DeleteTopics, CreatePartitions and DescribeConfigs aren't served, and topics aren't auto-created. A topic is a durable stream created through the Felix control plane, and its partition count is its shard count. In-memory streams aren't listed at all, since their offsets would point at records that are gone.

There's no compaction, and Kafka's retention settings don't apply. A tombstone is stored as an empty payload and deletes nothing.

Smaller gaps: no fetch sessions, no leader epochs, no topic ids (Fetch stops at v12), no SCRAM or OAUTHBEARER, a token checked once per connection, and no dotted namespaces. The duplicate window also differs. Kafka's docs say the broker “only retains at most 5 batches for each producer” (producer configs). Felix remembers where the last 64 batches landed, which for a Kafka producer means 64 records. A re-send older than that gets DUPLICATE_SEQUENCE_NUMBER, which clients count as delivered, but the answer has no offset.

Every behaviour here is tested with kcat 1.7.1 (librdkafka 1.8.2). The Java client is untested.

What we learned

Refusing loudly matters more than refusing early. The first prototype simply didn't answer FindCoordinator. kcat printed Waiting for group rebalance and hung there with no error (#498). Nobody holding that client could tell a missing feature from a slow cluster.

The protocol has no way to say “this broker has no coordinator”. So I tried codes until one did the right thing. librdkafka treats GROUP_AUTHORIZATION_FAILED as fatal for the group and passes the broker's message to the application, so kcat -G now exits with a sentence:

% ERROR: Consumer error: FindCoordinator response error: Felix has no Kafka consumer groups; assign partitions instead. See docs/kafka-compatibility.md

Transactions get the same treatment with TRANSACTIONAL_ID_AUTHORIZATION_FAILED.

The easy parts were easy because the shapes already matched. The spike that started this was 533 lines and an afternoon. What cost time was protocol detail, like answering a too-new ApiVersions in the v0 format, and Kafka's checksum being CRC-32C rather than CRC-32.

The thing worth spending time on was making the guarantee the log's, not the listener's. Kafka producers go through the same publish path, the same producer marks and the same replication as native clients. Failover, moves and restarts were already handled for Felix producers, so Kafka producers got that without separate code. The listener only had to translate.

Trying it

The Kafka compatibility guide has the full setup, including TLS and tokens. For a quick local try, the broker needs durable storage and the listener turned on. With the local Compose stack from the repo root, add a compose.override.yaml:

services:
  felix-broker:
    environment:
      FELIX_DURABLE_STORAGE_DIR: /var/lib/felix/data
      FELIX_KAFKA_LISTEN: 0.0.0.0:9092
      FELIX_KAFKA_ADVERTISE_ADDR: 127.0.0.1:9092
      FELIX_KAFKA_TLS: "false"
    ports:
      - "127.0.0.1:9092:9092"

Then start it, create a durable stream with three shards, and copy out the client token:

docker compose up -d
docker compose run --rm felixctl stream create clicks --shards 3
docker compose exec -T felix-broker cat /var/lib/felix/client.jwt > felix-dev.jwt

FELIX_KAFKA_TLS=false sends the token in clear text, so keep it on localhost. Now point kcat at it, with the tenant as the username and the token as the password:

K="-b 127.0.0.1:9092 -X security.protocol=SASL_PLAINTEXT -X sasl.mechanisms=PLAIN \
   -X sasl.username=dev -X sasl.password=$(cat felix-dev.jwt)"

kcat $K -L
printf 'one\ntwo\nthree\n' | kcat $K -P -t default.clicks -p 0 -X enable.idempotence=true
kcat $K -C -t default.clicks -o beginning -e -q -f '%p:%o:%s\n'

The offsets kcat prints are the shard's own. A Felix subscriber on clicks sees the same records at the same offsets. To see the refusal, try kcat $K -G mygroup default.clicks.

The reference, with every error mapping and config variable, is docs/kafka-compatibility.md.

docker compose up -d