bug: two brokers writing to the same partition can produce overlapping offsets

[klaudworks]

What happens

When two or more brokers receive produce requests for the same partition at the same time, they can both assign the same offset range to different messages. This means two different sets of messages end up in S3 claiming to be at the same offsets, and one silently overwrites or shadows the other.

Why it happens

Each broker keeps a local counter for the next offset to assign. This counter is set once when the broker first opens a partition (read from etcd), and then incremented in memory on every produce. The updated offset is only written back to etcd later, when the segment is flushed to S3 -- and that write is a simple overwrite, not a conditional update. So there is nothing stopping two brokers from reading the same starting offset, assigning overlapping ranges, and both writing their segments to S3.

The proxy, which sits in front of the brokers, does not look at which partition a produce request is targeting. It just round-robins client connections across available brokers. So if two producers (or even one producer reconnecting) land on different brokers and write to the same partition, the race can occur.

Proposed Fix

Assumption 1: We cannot properly fix this in the broker unless we introduce some slow etcd partition lease mechanism.
Assumption 2: We may need more than 1 proxy running in parallel o.w. we have a single point of failure and can't deploy this in HA mode.

Discarded Fix Proposal :

I evaluated a few solutions that would be based on deterministic hashing based on a broker list propagated via etcd but always ended up with edge cases that required a proper lease.

Fix Proposal via etcd Leases:

1. Brokers acquire an etcd lease per partition before writing. When a broker first receives a produce for a partition, it tries to grab a lease key in etcd (one round-trip). If it wins, it reads the current offset, creates the PartitionLog, and starts writing. All subsequent produces are purely local. If it loses, it rejects the request with NOT_LEADER_OR_FOLLOWER. When a broker dies, its leases expire automatically.
2. Proxies watch the lease keys to know where to route. The proxy watches partition ownership keys in etcd and routes produces to the broker that holds the lease. This is a best-effort optimization because the proxy's view can be briefly stale.
3. If a broker receives a produce for a partition it doesn't own, it rejects it. The proxy retries on another broker. This covers all edge cases: stale proxy routing table, lease just expired, new partition with no owner yet, scaling events. The broker's local lease check is the single source of truth. The proxy's routing avoids most retries, but the reject-and-retry fallback makes it bulletproof.

[klaudworks]

@novatechflow for this one I'll wait for your input before creating a PR.

[klaudworks]

Stable partition routing spares us the headaches to efficiently update offsets cross-broker and boosts fetch performance because fetches will always be served from the brokers cache instead of from s3. This leads to a lower boundary for e2e latency that is close to the time that it takes to upload single segment. Segment upload dominates everything else (which we want cause we can't avoid that)

[novatechflow]

I found that too. The main idea was that a broker is just a pod which can be killed when the assigned producer traffic stops. Using a broker per partition looks like the most pragmatic solution for a stable up- and downscale. We can mitigate overload with dynamic pod resources - meaning a broker starts with 2 CPU / 2GB RAM and when filled 80% resources scale up. Kubernetes 1.3x should support Pod Resource Autoscaler and VPA.

What do you think?

[klaudworks]

IMO your suggestion would be the goto when tackling the problem with a "cloud-native" mindset. Flink uses a similar approach for their operator when spawning task managers for jobs.

atm I see the following limitations:

huge resource usage: when you want to use it as an agentic communication layer, then I'd assume people may use lots of topics. Each topic using at least 1 pod with 2 CPU / 2GB RAM would lead to huge resource usage even when people would just use a single partition per topic. Let's assume they have just 10 topics and 2 partitions each to process them in parallel. We would already use 40 CPU / 40 GB RAM.
user experience: startup times of the broker, waiting for new cluster nodes to spawn via horizontal scaling, waiting for vertical pod scaling all adds up. Outside of the hyperscalers horizontal scaling of nodes may also take a few minutes or it may not even be available in the first place. Instead of simplifying the kafka management experience by not having to worry about backups, costs etc. we shift quite some burden and costs to the underlying k8s cluster. k8s is something that people struggle with anyways and many people have little control over their clusters.
my flink experience: was quite a hassle. At some point a user just decided to write small focussed jobs and spawned lots of task managers. That drained all cluster resources and was a waste of resources because most jobs used way less than the default resources assigned.

My current viewpoint: Managing a static amount of brokers would be the goto because your core vision of stateless s3 backed brokers remains valid. I think this is closer to the stated goal of having simple, cheap Kafka that covers most use cases.

Introducing an etcd-based lease mechanism definitely introduces complexity but it is complexity that we can properly test and hide from the user. I'd rather have the complexity in the internals than exposed to the user. The user will only have to worry about adding or removing replicas when they are not happy with their current performance.

@novatechflow Is there something I overlook? If you wanna discuss this, we can also have a quick chat about it today just hmu on linkedin. Just added you.

PS: I recently used test containers a lot. This could provide some more safety around etcd and multibroker stuff. In another project I built 500 integration tests which used keycloak, ceph, flink, kafka, kubernetes mock api server and they still run in < 3min on my M1 Macbook.
PSPS: I also see a future where this can be used as a complete drop-in replacement to kafka by introducing a limited WAL and disk-based caching of topics to achieve kafka like latencies.

[novatechflow]

Its also valid - we start with 3 broker, scale up, but never less then 3 to have stable quorum.

[klaudworks]

Does the number of brokers actually matter? In my current understanding it only matters for etcd Raft consensus.

Regarding quorum: fyi I actually reduced min. etcd nodes to 1 in this already merged PR https://github.com/KafScale/platform/pull/119/changes. My reasoning was that it should be 3 etcd nodes for HA but 1 is fine for testing. Proper check would probably be that count(etcdNodes) modulo 2 == 1 to have an uneven number of etcd nodes for quorum.