In a nutshell we compare multi-partition write (BaseWritesBench) with atomic multi-partition write (TxWritesBench) and naive streaming with transactional streaming (TxStreamingBench).

Transactional writes have 30x-50x overhead compared to baseline.

Transactional streaming processing has 40x-120x overhead compared to baseline.

BaseWritesBench sends two messages to two topics asynchronously (one to each topic, using KafkaProducer.send) and then waits until the returned futures resolve. It does it in a loop and records how long each iteration took.

BaseStreamingBench fills a source topic, reads the source topic in batches then for each record in a batch it sends a message to a target topic (KafkaProducer.send), sends an offset to a consumer group (KafkaConsumer.commitSync) and waits until the returned future resolves. The test measures only the write path (send and commitSync) and doesn't include the fetch part.

TxWritesBench is similar to BaseWritesBench with an exception it wraps each iteration in a transaction: the test starts a transaction, sends two messages to two topics asynchronous (one to each topic, using KafkaProducer.send), commits the transaction and waits until the returned futures resolve.

TxStreamingBench is similar to BaseStreamingBench: fills a source topic, reads the source topic in batches then for each record in a batch it starts a transactions, sends a message to a target topic (KafkaProducer.send), sends an offset to a consumer group (KafkaProducer.sendOffsetsToTransaction) and waits until the returned future resolves. The test measures only the write path (send and sendOffsetsToTransaction) and doesn't include the fetch part.

Usually performance tests compensate for the coordinated omission. But we don't do it here because we measure different things. Avoiding coordinated omission matters when we want to answer to a question how a system behaves with a given load (on average a system has X tps). While we want to measure the theoretical limitations of the transaction protocol (a system can't do better than Y tps) so we execute transactions in isolation, one after another over and over again.

Spawn a 3-nodes (i3.large, same AZ) Kafka (2.7) cluster backed by NVMe (/dev/nvme0n1) drive using XFS file system. With settings based on settings/zoo.cfg and settings/kafka.properties

(don't forget to open the intra-cluster ports: 2888, 8888, 33145, 3888, 9092)

Spawn a control node with JDK, Maven, Kafka (with environment variable KAFKA pointed to it).

Clone this repo & build the tests on the control node:

git clone
cd kafka-tx-performance
pushd tests
mvn clean dependency:copy-dependencies package
popd

Create the target topics:

$KAFKA/bin/kafka-topics.sh --create --topic topic1 --partitions 1 --replication-factor 3 --bootstrap-server NODE1_IP:9092
$KAFKA/bin/kafka-topics.sh --create --topic topic2 --partitions 1 --replication-factor 3 --bootstrap-server NODE1_IP:9092

Run the experiments (before each experiment kill the kafka and zookeeper processes, wipe the data directory, restart the processes and recreate the topics):

KAFKA_CONNECTION=NODE1_IP:9092 WARMUP=1000 TXES=1000 MEASURES=kafka.base.writes.log java -cp $(pwd)/tests/target/performance-1.0-SNAPSHOT.jar:$(pwd)/tests/target/dependency/* io.vectorized.tests.BaseWritesBench
 
KAFKA_CONNECTION=NODE1_IP:9092 WARMUP=1000 TXES=1000 MEASURES=kafka.base.streaming.log java -cp $(pwd)/tests/target/performance-1.0-SNAPSHOT.jar:$(pwd)/tests/target/dependency/* io.vectorized.tests.BaseStreamingBench
 
KAFKA_CONNECTION=NODE1_IP:9092 WARMUP=1000 TXES=1000 MEASURES=kafka.tx.writes.log java -cp $(pwd)/tests/target/performance-1.0-SNAPSHOT.jar:$(pwd)/tests/target/dependency/* io.vectorized.tests.TxWritesBench
 
KAFKA_CONNECTION=NODE1_IP:9092 WARMUP=1000 TXES=1000 MEASURES=kafka.tx.streaming.log java -cp $(pwd)/tests/target/performance-1.0-SNAPSHOT.jar:$(pwd)/tests/target/dependency/* io.vectorized.tests.TxStreamingBench