The LogDrive: Flexible Composition Through Abstraction in Shared Logs

In July, several Confluent colleagues and I published The LogDrive: Composable Durability for Cloud-Based Shared Logs. The paper evolves the concepts of Virtual Consensus in Delos by decomposing the Loglet abstraction into two additional abstractions that aid Loglet composition. This post describes these new abstractions in the context of Virtual Consensus, focusing on how they support RAID-like composability for shared logs.

The ideas of the paper came out of a research project led by Mahesh Balakrishnan, Gardner Vickers, and Lucas Bradstreet. The project itself was to build a Kafka-on-S3 service to run in Confluent Cloud and the project ultimately became K2 or Kora 2, the next evolution of the Kora engine (which powers Confluent Cloud’s Kafka service). The first product built on K2 was Freight Clusters (Kafka on S3). The research that led to the LogDrive paper originated in Conflux, the scalable metadata service in K2 which acts as sequencer and metadata database for the fleet of leaderless brokers, akin to WarpStream’s agents.

Apache Kafka performance #1 - linger.ms

This is the first in an ongoing ad-hoc series of posts on Apache Kafka performance. I have no general direction, I’ll just share interesting insights based on the performance testing I do on Apache Kafka.

Recently I was curious to see if there was any general performance improvement since Kafka 3.X. So I ran a suite of benchmarks with Dimster against 3.7.2 and 4.3.0. I saw two common patterns:

Pattern 1: Low load benchmarks showed that end-to-end latency was higher with Kafka 4.3 compared to 3.7.2. The following is a 45 minute no-record-key workload of 5000 record/s, 20 topics (120 partitions), fan-out 2 (240 consumers), full TLS, on 3 brokers each allocated 8 SMT CPUs in k8s (on my Threadripper 9980X).

1BRC on a Threadripper 9980X

esterday I published some benchmarks of Hardwood 1.0 on my Threadripper. Someone suggested I run the One Billion Row Challenge too, to see how it does, so here it is!

Gunnar Morling ran the original benchmarks on an EPYC 7502P, Zen 2, 32 cores with 128 GB of RAM. The official challenge was on 8 cores (sequentially chosen) plus a bonus of all 32 cores.

I chose to run the benchmark using 9 contenders from the published 8 and 32 core results. The 9 contenders I ran were thomaswue, artsiomkorzun, jerrinot, serkan-ozal, abeobk, stephenvonworley, royvanrijn, mtopolnik, yavuztas.

Benchmarking Hardwood 1.0 on a Threadripper 9980X

Hardwood is a minimal-dependency Java library for reading Parquet files. It currently has row-reader and columnar-reader APIs, with Parquet writing planned for the future.

Gunnar Morling, Hardwood’s author, published some initial benchmarks in the v1.0 announcement, comparing Hardwood’s row and column readers against Parquet Java. Those benchmarks measured read speed against already-downloaded Parquet files. 

Gunnar’s benchmarks ran on an m7i.2xlarge, with 8 vCPUs / 4 physical cores. Each test used three variants:

  • Hardwood with decoder threads = Runtime.getRuntime().availableProcessors(), which equals 8

  • Hardwood pinned to one CPU thread with taskset

  • Parquet Java, single-threaded

I was curious how the same benchmarks would look on my Threadripper 9980X: 64 cores / 128 threads, with 256 GB ECC DDR5. I modified Gunnar’s benchmark code to also test Hardwood with fixed decoder-thread counts: 1, 4, and 8.

Kafka Share Groups - Pathological fetch waits with record_limit

In this post we’re going to see how share.acquire.mode=record_limit combined with fewer consumers than partitions and various cases of “partition skew” can result in subpar performance with share groups. 

I stumbled on these issues when running large sets of dimensional tests with Dimster’s explore-limits mode, which finds the highest sustainable throughput while staying within a target end-to-end latency target. There was a specific subset of the tests that explore-limits mode would consistently fail to complete, and they all happened to be with record_limit and a consumer count lower than the partition count. In this test, we’ll understand why Dimster had such a hard time with this combination.

,

Can We Agree on a Storage/Workload Architecture Taxonomy?

,

The lines between transactional systems, analytical systems, hybrid systems, and shared storage architectures are getting blurry. This post proposes a small taxonomy for describing the different ways systems, workloads, storage tiers, visibility, and durable copies relate to each other.

OLTP, OLAP, HTAP, and now LTAP.

We can think of the first two as two types of workload which have specialized query engines and storage systems to support them. OLTP such as the RDBMS like Postgres and MySQL use row-based storage engines. OLAP, such as Clickhouse, cloud data warehouse and the lakehouse use column-based storage.

HTAP is a hybrid workload system: one system -> both transactional and analytical workloads. The HTAP system therefore has specialized storage and specialized query engine to stitch together the row-based and columnar data.

So far, we’re dealing with a single system. A Postgres (OLTP), a Clickhouse (OLAP), a SingleStore or TiDB (HTAP).

So what is LTAP?

Raise the ambition threshold

“Perfection is finally attained not when there is no longer anything to add, but when there is no longer anything to take away.” — Antoine de Saint-Exupéry

AI gives us an unprecedented ability to add. The danger is that we begin to mistake accumulation for value.

Delivery is only the beginning (or be mindful of catabolic collapse)

Every new system and feature adds obligations: it must be operated, secured, monitored, documented, integrated, upgraded and eventually replaced or retired. Hackers love a juicy target, even if it’s that half-forgotten service that people are unsure whether it’s safe to turn off or not. If we respond to “cheaper” software creation by producing far more software, we may accumulate obligations faster than we acquire the capacity to discharge them. Under the weight of the proliferation of software, the organization starts to sacrifice its ability to build what it will need next to react effectively to changing market conditions and opportunities.

This is the dynamic described by catabolic collapse.

Kafka Share Groups and Parallelizing Consumption - Part 3: Client-local parallelism

In the last post Broker-Visible vs Client-Local Parallelism we looked at two ways of scaling Kafka consumption. The final unit of parallelism can be visible to the broker, as consumers, or it can be local to the client, as threads, virtual threads, async tasks, or some other execution mechanism hidden behind a smaller number of consumers. 

Broker-visible parallelism is simple to reason about: if each consumer processes records serially, we add more consumers to increase parallelism. But each consumer adds overhead to the brokers: broker-side protocol state, TCP connections, group membership, fetch state, and participation in the consumer or share group protocol. With long processing times and/or high throughput, the required number of parallel workers can easily exceed what is practical to model as broker-visible consumers.

That is where client-local parallelism becomes important. Instead of scaling by adding more consumers, each consumer application can poll records and process them concurrently inside the client. This allows a smaller number of Kafka consumers to drive a much larger amount of parallel work.

In this post, we’ll compare client-local parallelism with consumer groups and share groups using the Apache Kafka clients, by way of Dimster, the benchmarking tool used throughout this series. Dimster uses the official Apache Kafka clients under the hood. The main comparison is between two styles of client-local parallelism: blocking and continuous styles.

,

Broker-Visible vs Client-Local Parallelism

,

This post is a little side-quest from my “Kafka Share Groups and Parallelizing Consumption” series.

My “Kafka Share Groups and Parallelizing Consumption” series (part 1, part 2) has been laser focused on how different configurations and behaviors affect parallel consumption in share groups. So far I’ve shown that you most definitely can hold share groups wrong. You could quite easily and inadvertently create a work queue and with the right combination of things going against you, see a small number of consumers dominate, leaving most consumers starved of messages. All the while lag builds and builds. You need to know the settings and what they do.

But it’s worth asking the question: is parallelizing consumption what share groups are for?

Kafka Share Groups and Parallelizing Consumption - Part 2: Producer Batches and share.acquire.mode

In the last post we used simulated consumer processing time to reveal how important it is to set an appropriate value for max.poll.records. The rule of thumb was a value somewhat lower than:

group.share.partition.max.record.locks / number of consumers per partition

But there’s more to parallel consumption than max.poll.records. The size of producer batches also plays a role when using the default share.acquire.mode (batch_optimized).