Page History

...

1. High write load: it should be okay if consumers want to commit after every message if they need to do that (it will be slower, but the system shouldn't collapse). We need to be able to horizontally scale offset writes.
2. Durability: once an offset is written it can't be lost even if a server crashes or disappears.
3. Consistent reads: all reads must return the last written value--no "eventual consistency" can be apparent.
4. Small data size--potentially in memory. A consumer-group/topic/partition/offset tuple should be no more than 64 bytes in memory even with all the inefficiencies of jvm object overhead assuming the strings are interned and a compact lookup structure. So each 1GB of memory could potentially support ~16 million entries.
5. Transactionality across updates for multiple topic/partitions. This is perhaps not a hard requirement but would be nice to have. What I mean by this is that if a consumer subscribes to two partitions and does issues a commit of its position for request for its offset in both partitions that commit should either fail for both or succeed for both, since handling a partial success could be fairly complex. This is not true in our zk usage today fwiw.

So here is a strawman proposal. We make use of the keyed topic support we are adding and store the offset commits in Kafka as a topic. The implementation of OffsetCommit would just be publishing a message to the "offset-commit" topic. The cleaner process described in the keyed topic support wiki would do the deduplication to remove obsolete messages from the log periodically. This would support very high write throughput with a partitioning model to scale load as well as giving durability and fault tolerance guarantee from Kafka replication. This obviously would not support fast reads, though. To support fast reads the brokers would create an "index" of this topic in memory by also populating a hash map or other in-memory store. On restart the broker would restore this hash map from the offset-commit topic's contents.

Open Questions

There are a number of ways to handle partitioning of this "offset-commit" topic, each with different trade offs.

My proposal would be that we partition by consumer group, and that each message in the topic contain the complete set of updates sent in one request. This would give us the transactionality since our guarantees today are always at the message level. The broker would explode each of these multi-partition updates into multiple hash map entries. The partitioning would be handled by the broker. That is, the client could issue its commit request to any broker which would in turn issue a send() call to the broker responsible for the partition containing that consumer group. Likewise the get offset call would be proxied to the correct broker (unless by chance it landed on the correct broker).

The advantage of this approach is that updates are transactional and also that the client can be stupid and just direct its requests to whatever broker it happens to be connected to.

The downside of this particular partitioning arrangement is the "two hops" necessary. A savy client could optimize this away by using the same hash the broker does on the consumer group to chose the broker to make its request to.How to partition the topic?