How to Reduce Cloud Cost by 99% for EDA Kafka Applications

While the cloud offers great convenience and flexibility, the operational cost for applications deployed in the cloud can sometimes be significant. This article shows a way to substantially reduce operating costs in latency-sensitive Event-Driven Architecture (EDA) Java applications by migrating from Kafka to Chronicle Queue open-source, a more resource-efficient and lower-latency queue implementation.

What is EDA?

An EDA application is a distributed application where events (in the form of messages or DTOs) are produced, detected, consumed, and reacted to. Distributed means it might run on different machines or the same machine but in separate processes or threads. The latter concept is used in this article whereby messages are persisted to queues.

Setting the Scene

Suppose that we have an EDA application with a chain of five services and where we have a requirement that 99.9% of the messages sent from the first producer to the last consumer should have a latency less than 100 ms at a message rate of 1,000 messages per second.

Figure 1, The five Services and the Benchmark are interconnected by six topics/queues.

In other words, the time it takes from sending a message (ie using topic 0) by the Benchmark thread to when a resulting message is received by the Benchmark thread again (ie via topic 5) is only allowed to be higher than 100 ms for on average one messages out of every 1,000 messages which are sent every second.

The messages used in this article are simple. They contain a long nanosecond timestamp holding the initial timestamp when a message is first posted via topic 0 and an int value that is increased by one each time the message is propagated from one service to the next (this value is not actually used but illustrates a rudimentary service logic). When a message arrives back at the Benchmark thread, the current nanotime is compared with the original nanotime in the initial message sent on topic 0 to allow the calculation of the total latency across the entire service chain. The latency samples are then subsequently fed into a histogram for later analysis.

As can be seen in Figure 1 above, the number of topics/queues is equal to the number of services plus one. Hence, there are six topics/queues because there are five services.

The Question

The question in this article is: How many instances of these chains can we set up on a given hardware and still meet the latency requirement? Or, to rephrase it, how many of these applications can we run and still pay the same price for the hardware used?

Default Setup

In this article, I have elected to use Apache Kafka because it is one of the most common queue types used in the market. I have also selected Chronicle Queue due to its ability to provide low latency and resource efficiency. 

Both Kafka and Chronicle Queue have several configurable options, including replicating data across several servers. In this article, a single non-replicated queue will be used. For performance reasons, the Kafka broker will be run on the same machine as the services, allowing the use of the local loopback network interface.

The KafkaProducer instances are configured to be optimised for low latency (eg setting “acks=1”), and so are the KafkaConsumer instances.

The Chronicle Queue instances are created using the default setup with no explicit optimisation. Hence, the more advanced performance features in Chronicle Queue like CPU-core pinning and busy spin-waiting are not used.

Kafka

Apache Kafka is an open-source distributed event streaming platform for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications used extensively in various EDA applications, especially when several information sources residing in different locations are to be aggregated and consumed. 

In this benchmark, each test instance will create six distinct Kafka topics, and they are named topicXXXX0, topicXXXX1, … , topicXXXX5 where XXXXX is a random number.

Chronicle Queue

Open-source Chronicle Queue is a persisted low-latency messaging framework for high-performance and critical applications. Interestingly, Chronicle Queue uses off-heap memory and memory-mapping to reduce memory pressure and garbage collection impacts, making the product popular within the fintech area where deterministic low latency messaging is crucial.

In this other benchmark, each test instance will create six Chronicle Queue instances, named topicXXXX0, topicXXXX1, … , topicXXXX5 where XXXXX is a random number.

Code

The inner loops for the two different service thread implementations are shown below. They both poll their input queue until being ordered to shut down and, if there are no messages, they will wait for one-eighth of the expected inter-message time before a new attempt is made.

Here is the code:

Kafka

while (!shutDown.get()) {

    ConsumerRecords<Integer, Long> records = 

            inQ.poll(Duration.ofNanos(INTER_MESSAGE_TIME_NS / 8));

    for (ConsumerRecord<Integer, Long> record : records) {

        long beginTimeNs = record.value();

        int value = record.key();

        outQ.send(new ProducerRecord<>(topic, value + 1, beginTimeNs));

    }

}

Using the record key() to carry an int value might be a bit unorthodox but allows us to improve performance and simplify the code.

Chronicle Queue

while (!shutDown.get()) {

    try (final DocumentContext rdc = tailer.readingDocument()) {

        if (rdc.isPresent()) {

            ValueIn valueIn = rdc.wire().getValueIn();

            long beginTime = valueIn.readLong();

            int value = valueIn.readInt();

            try (final DocumentContext wdc =

                         appender.writingDocument()) {

                final ValueOut valueOut = wdc.wire().getValueOut();

                valueOut.writeLong(beginTime);

                valueOut.writeInt(value + 1);

            }

        } else {

            LockSupport.parkNanos(INTER_MESSAGE_TIME_NS / 8);

        }

    }

}

Benchmarks

The benchmarks had an initial warmup phase during which the JVM’s C2 compiler profiled and compiled code for much better performance. The sampling results from the warmup period were discarded.

More and more test instances were started manually (each with its own five services) until the latency requirements could not be fulfilled anymore. Whilst running the benchmarks, the CPU utilisation was also observed for all instances using the “top” command and averaged over a few seconds.

The benchmarks did not take coordinated omission into account and were run on  Ubuntu Linux (5.11.0-49-generic) with AMD Ryzen 9 5950X 16-Core Processors at 3.4 GHz with 64 GB RAM where the applications were run on the isolated cores 2-8 (7 CPU cores in total) and queues were persisted to a 1 TB NVMe flash device . OpenJDK 11 (11.0.14.1) was used.

All latency figures are given in ms, 99% means 99-percentile and 99.9% means 99.9-percentile.

Kafka

The Kafka broker and the benchmarks were all run using the prefix “taskset -c 2-8” followed by the respective command (eg taskset -c 2-8 mvn exec:java@Kafka). The following results were obtained for Kafka:

Instances	Median Latency	99%	99.9%	CPU Utilisation
1	0.9	19	30	670%
2	16	72	106 (*)	700% (saturated)

Table 1, Shows Kafka instances vs latencies and CPU utilisation.

(*) Over 100 ms on the 99.9-percentile.

As can be seen, only one instance of the EDA system could be run simultaneously. Running two instances increased the 99.9-percentile, so it exceeded the limit of 100 ms. The instances and the Kafka broker quickly saturated the available CPU resources.

Here is a snapshot from the output from the “top” command when running two instance and a broker (pid 3132946):

3134979 per.min+  20   0   20.5g   1.6g  20508 S 319.6   2.6  60:27.40 java                                                                            

3142126 per.min+  20   0   20.5g   1.6g  20300 S 296.3   2.5  19:36.17 java                                                                            

3132946 per.min+  20   0   11.3g   1.0g  22056 S  73.8   1.6   9:22.42 java

Chronicle Queue

The benchmarks were run using the command “taskset -c 2-8 mvn exec:java@ChronicleQueue” and following results were obtained:

Instances	Median Latency	99%	99.9%	CPU Utilisation
1	0.5	0.8	0.9	5.2%
10	0.5	0.9	0.9	79%
25	0.5	0.9	3.6	180%
50	0.5	0.9	5.0	425%
100	1.0	5	20	700% (saturated)
150	2.0	7	53	700% (saturated)
200	3.1	9	59	700% (saturated)
250	4.8	12	62	700% (saturated)
375	8.7	23	75	700% (saturated)
500	11	36	96	700% (saturated)

Table 2, Shows Chronicle Queue instances vs latencies and CPU utilisation.

The sheer efficiency of Chronicle Queue becomes apparent in these benchmarks when 500 instances can be run at the same time meaning we handle 3,000 simultaneous queues and 3,000,000 messages per second on just 7 cores at less than 100 ms delay at the 99.9-percentile.

Comparison

Here is a chart showing the number of instances vs the 99.9-percentile for the two different queue types:

Chart 1, Shows Instances vs latencies in ms for the 99.9-percentile.

As can be seen, the curve for Kafka goes from 30 ms to 106 ms in just one step so the latency growth for Kafka looks like a wall in this scale.

Conclusion
About four hundred times more applications can be run on the same hardware if a switch is made from Kafka to Chronicle Queue for specific latency-sensitive EDA applications. 

About four hundred times more applications corresponds to a potential of reducing cloud or hardware costs by about 99.8% as illustrated in Char 2 below (less is better). In fact, the cost can barely be seen at all in the scale used:

Chart 2, Shows normalised cost vs queue type (less is better)

Resources

Open-source Apache Kafka

Open-source Chronicle Queue

 

Did You Know the Fastest Way of Serializing a Java Field is not Serializing it at All? 

This article elaborates on different ways of serializing Java objects and benchmarks performance for the variants. Read this article and become aware of different ways to improve Java serialization performance.

In a previous article about open-source Chronicle Queue, there was some benchmarking and method profiling indicating that the speed of serialization had a significant impact on execution performance. After all, this is only to be expected as Chronicle Queue (and other persisted queue libraries) must convert Java objects located on the heap to binary data which is subsequently stored in files. Even for the most internally efficient libraries, this inevitable serialization procedure will largely dictate performance.

Data Transfer Object

In this article, we will use a Data Transfer Object (hereafter DTO) named  MarketData which contains financial information with a relatively large number of fields. The same principles apply to other DTOs in any other business area.

abstract class MarketData extends SelfDescribingMarshallable {

    long securityId;

    long time;

    // bid and ask quantities

    double bidQty0, bidQty1, bidQty2, bidQty3;

    double askQty0, askQty1, askQty2, askQty3;

    // bid and ask prices

    double bidPrice0, bidPrice1, bidPrice2, bidPrice3;

    double askPrice0, askPrice1, askPrice2, askPrice3;

    // Getters and setters not shown for clarity

}

Default Serialization

Java’s Serializable marker interface provides a default way to serialize Java objects to/from binary format, usually via the ObjectOutputStream and ObjectInputStream classes. The default way (whereby the magic writeObject() and readObject() are not explicitly declared) entails reflecting over an object's non-transient fields and reading/writing them one by one, which can be a relatively costly operation.

Chronicle Queue can work with Serializable objects but also provides a similar, but faster and more space-efficient way to serialize data via the abstract class SelfDescribingMarshallable. Akin to Serializable objects, this relies on reflection but comes with substantially less overhead in terms of payload, CPU cycles, and garbage.

Default serialization often comprises the steps of:

• Identifying the non-transient fields using reflection

• Reading/writing the identified non-transient field values using reflection

• Writing/reading the field values to a target format (eg binary format)

The identification of non-transient fields can be cached, eliminating this step to improve performance.

Here is an example of a class using default serialization:

public final class DefaultMarketData extends MarketData {}

As can be seen, the class does not add anything over its base class and so it will use default serialization as transitively provided by SelfDescribingMarshallable.

Explicit Serialization

Classes implementing Serializable can elect to implement two magic private (sic!) methods whereby these methods will be invoked instead of resorting to default serialization.

This provides full control of the serialization process and allows fields to be read using custom code rather than via reflection which will improve performance. A drawback with this method is that if a field is added to the class, then the corresponding logic must be added in the two magic methods above or else the new field will not participate in serialization. Another problem is that private methods are invoked by external classes. This is a fundamental violation of encapsulation.

SelfDescribingMarshallable classes work in a similar fashion but thankfully it does not rely on magic methods and invoking private methods externally. A SelfDescribingMarshallable class provides two fundamentally different concepts of serializing: one via an intermediary Chronicle Wire open-source (which can be binary, text, YAML, JSON, etc) providing flexibility and one implicitly binary providing high performance. We will take a closer look at the latter one in the sections below.

Here is an example of a class using explicit serialization whereby public methods in implementing interfaces are explicitly declared:

public final class ExplicitMarketData extends MarketData {

    @Override

    public void readMarshallable(BytesIn bytes) {

        securityId = bytes.readLong();

        time = bytes.readLong();

        bidQty0 = bytes.readDouble();

        bidQty1 = bytes.readDouble();

        bidQty2 = bytes.readDouble();

        bidQty3 = bytes.readDouble();

        askQty0 = bytes.readDouble();

        askQty1 = bytes.readDouble();

        askQty2 = bytes.readDouble();

        askQty3 = bytes.readDouble();

        bidPrice0 = bytes.readDouble();

        bidPrice1 = bytes.readDouble();

        bidPrice2 = bytes.readDouble();

        bidPrice3 = bytes.readDouble();

        askPrice0 = bytes.readDouble();

        askPrice1 = bytes.readDouble();

        askPrice2 = bytes.readDouble();

        askPrice3 = bytes.readDouble();

    }

    @Override

    public void writeMarshallable(BytesOut bytes) {

        bytes.writeLong(securityId);

        bytes.writeLong(time);

        bytes.writeDouble(bidQty0);

        bytes.writeDouble(bidQty1);

        bytes.writeDouble(bidQty2);

        bytes.writeDouble(bidQty3);

        bytes.writeDouble(askQty0);

        bytes.writeDouble(askQty1);

        bytes.writeDouble(askQty2);

        bytes.writeDouble(askQty3);

        bytes.writeDouble(bidPrice0);

        bytes.writeDouble(bidPrice1);

        bytes.writeDouble(bidPrice2);

        bytes.writeDouble(bidPrice3);

        bytes.writeDouble(askPrice0);

        bytes.writeDouble(askPrice1);

        bytes.writeDouble(askPrice2);

        bytes.writeDouble(askPrice3);

    }

}

It can be concluded that this scheme relies on reading or writing each field explicitly and directly, eliminating the need to resort to slower reflection. Care must be taken to ensure fields are referenced in a consistent order and class fields must also be added to the methods above. 

Trivially Copyable Serialization

The concept of Trivially Copyable Java Objects is derived from and inspired by C++. 

As can be seen, the MarketData class above contains only primitive fields. In other words, there are no reference fields like String, List or the like. This means that when the JVM lays out the fields in memory, field values can be put adjacent to one another. The way fields are laid out is not specified in the Java standard which allows for individual JVM implementation optimizations. 

Many JVMs will sort primitive class fields in descending field size order and lay them out in succession. This has the advantage that read and write operations can be performed on even primitive type boundaries. Applying this scheme on the  ExplicitMarketData for example will result in the long time field being laid out first and, assuming we have the initial field space 64-bit aligned, allows the field to be accessed on an even 64-bit boundary. Next, the int securityId might be laid out, allowing it and all the other 32-bit fields to be accessed on an even 32-bit boundary. 

Imagine instead if an initial byte field were initially laid out, then subsequent larger fields would have to be accessed on uneven field boundaries. This would add a performance overhead for some operations, and would indeed prevent a small set of operations from being performed at all (eg unaligned CAS operations on the ARM architecture).

How is this relevant to high-performance serialization? Well, as it turns out, it is possible to access an object’s field memory region directly via Unsafe and use memcpy to directly copy the fields in one single sweep to memory or to a memory-mapped file. This effectively bypasses individual field access and replaces, in the example above, the many individual field accesses with a single bulk operation. 

The way this can be done in a correct, convenient, reasonably portable and safe way is outside the scope of this article. Luckily, this feature is readily available in Chronicle Queue, open-source Chronicle Bytes and other similar products out-of-the-box.

Here is an example of a class using trivially copyable serialization:

import static net.openhft.chronicle.bytes.BytesUtil.*;

public final class TriviallyCopyableMarketData extends MarketData {

    static final int START = 

            triviallyCopyableStart(TriviallyCopyableMarketData.class);

    

    static final int LENGTH = 

            triviallyCopyableLength(TriviallyCopyableMarketData.class);

    @Override

    public void readMarshallable(BytesIn bytes) {

        bytes.unsafeReadObject(this, START, LENGTH);

    }

    @Override

    public void writeMarshallable(BytesOut bytes) {

        bytes.unsafeWriteObject(this, START, LENGTH);

    }

}

This pattern lends itself well to scenarios where the DTO is reused. Fundamentally, It relies on invoking Unsafe under the covers for improved performance.

Benchmarks

Using JMH, serialization performance was assessed for the various serialization alternatives above using this class:

@State(Scope.Benchmark)

@BenchmarkMode(Mode.AverageTime)

@OutputTimeUnit(NANOSECONDS)

@Fork(value = 1, warmups = 1)

@Warmup(iterations = 5, time = 200, timeUnit = MILLISECONDS)

@Measurement(iterations = 5, time = 500, timeUnit = MILLISECONDS)

public class BenchmarkRunner {

    private final MarketData defaultMarketData = new DefaultMarketData();

    private final MarketData explicitMarketData = new ExplicitMarketData();

    private final MarketData triviallyCopyableMarketData = new TriviallyCopyableMarketData();

    private final Bytes<Void> toBytes = Bytes.allocateElasticDirect();

    private final Bytes<Void> fromBytesDefault = Bytes.allocateElasticDirect();

    private final Bytes<Void> fromBytesExplicit = Bytes.allocateElasticDirect();

    private final Bytes<Void> fromBytesTriviallyCopyable = Bytes.allocateElasticDirect();

    public BenchmarkRunner() {

        defaultMarketData.writeMarshallable(fromBytesDefault);

        explicitMarketData.writeMarshallable(fromBytesExplicit);

        triviallyCopyableMarketData.writeMarshallable(fromBytesTriviallyCopyable);

    }

    public static void main(String[] args) throws Exception {

        org.openjdk.jmh.Main.main(args);

    }

    @Benchmark

    public void defaultWrite() {

        toBytes.writePosition(0);

        defaultMarketData.writeMarshallable(toBytes);

    }

    @Benchmark

    public void defaultRead() {

        fromBytesDefault.readPosition(0);

        defaultMarketData.readMarshallable(fromBytesDefault);

    }

    @Benchmark

    public void explicitWrite() {

        toBytes.writePosition(0);

        explicitMarketData.writeMarshallable(toBytes);

    }

    @Benchmark

    public void explicitRead() {

        fromBytesExplicit.readPosition(0);

        explicitMarketData.readMarshallable(fromBytesExplicit);

    }

    @Benchmark

    public void trivialWrite() {

        toBytes.writePosition(0);

        triviallyCopyableMarketData.writeMarshallable(toBytes);

    }

    @Benchmark

    public void trivialRead() {

        fromBytesTriviallyCopyable.readPosition(0);

        triviallyCopyableMarketData.readMarshallable(fromBytesTriviallyCopyable);

    }

}

This produced the following output on a MacBook Pro (16-inch, 2019) with 2.3 GHz 8-Core Intel Core i9 CPU under JDK 1.8.0_312, OpenJDK 64-Bit Server VM, 25.312-b07:

Benchmark                      Mode  Cnt   Score   Error  Units

BenchmarkRunner.defaultRead    avgt    5  88.772 ± 1.766  ns/op

BenchmarkRunner.defaultWrite   avgt    5  90.679 ± 2.923  ns/op

BenchmarkRunner.explicitRead   avgt    5  32.419 ± 2.673  ns/op

BenchmarkRunner.explicitWrite  avgt    5  38.048 ± 0.778  ns/op

BenchmarkRunner.trivialRead    avgt    5   7.437 ± 0.339  ns/op

BenchmarkRunner.trivialWrite   avgt    5   7.911 ± 0.431  ns/op

Using the various MarketData variants, explicit serialization is more than two times faster than default serialization. Trivially copyable serialization is four times faster than explicit serialization and more than ten times faster than default serialization as illustrated in the graph below (lower is better):

More fields generally favour trivially copyable serialization over explicit serialization. Experience shows break-even is reached at around six fields in many cases. 

Interestingly, the concept of trivially copyable can be extended to hold data normally stored in reference fields such as a String or an array field. This will provide even more relative performance increase for such classes.  Contact the Chronicle team if you want to learn more,

Why Does it Matter?

Serialization is a fundamental feature of externalizing DTOs to persistent queues, sending them over the wire or putting them in an off-heap Map and otherwise handling DTOs outside the Java heap. Such data-intensive applications will almost always gain performance and experience reduced latencies when the underlying serialization performance is improved.

Resources

Chronicle Queue (open-source)

GitHub Chronicle Bytes (open-source)