Example

For example, a stream

should produce

Which tells us that “Sensor 1” was “off” from 15:45:00 for 30 seconds and “on” from 15:45:30 for 60seconds.

Note that the second “off” reading produced an intermediate result.

Design decisions

Duplicate readings of the same state generate intermediate results, and delayed readings (timestampspreceding previously seen values) are treated as errors.

These are deliberate choices and can easily be changed.

Implementation of Business Logic

Care has been taken to keep the business logic independent of implementation details likeserialization formats.

The data model is in themodel directory, thebusiness logic inlogic.

Thetests test the topology with ninedifferent formats,Protocol Buffers,JSON,Apache Avro, theConfluent variants of these three,XML,Apache Thrift andAmazon Ion. Different, random combinations of input, result, andstate store formats are tested.

While this abstraction might not be necessary in practice, it demonstrates two important designconsiderations:

• The business logic should only depend on a data model, not capabilities of the serializationmechanism.

We can simply useDuration::between,which is a simple call and easy to understand and test, instead of cluttering our logic withconversions and unnecessary error-prone calculations.

• The choice of (de-)serializers should depend on the requirements, not on what is just at hand.

While internal processing pipelines tend (but don’t have) to use one serialization mechanism, it isperfectly valid and a good design decision to use different mechanisms for parts interfacing withexternal components.

Since the business logic is independent of the serialization mechanism, changing it is simple andusually does not require retesting.

By refactoring the business logic to depend only on an abstract store, we speed up testing by afactor of seven(bazel test //src/test/java/com/fillmore_labs/kafka/sensors/logic:all vs.bazel test //src/test/java/com/fillmore_labs/kafka/sensors/topology:all), which demonstrates a potentialfor improvement in development speed and testability.

Prerequisites

You needBazelisk installed, see alsoInstalling Bazel using Bazelisk.

Tests

To run all tests, use

To run a single test, use

The tests run with an embedded Kafka and mock schema registry, when necessary.

Main App

The main app needs Kafka running atlocalhost, port 9092 (seeapplication.yaml). There is a script doing that:

When Kafka has finished starting, create the topics in a different terminal:

Now start the main app:

Open another terminal to watch the results:

Publish sensor values:

Benchmark

Run theJMH microbenchmarks with

Compare deserialization of two formats:

Generate a flame graph for detailed analysis:

Run the latest image on your Kubernetes cluster:

Mapping

As noted inImplementation of Business Logic the business login is independent of theserialization, in the spirit of hexagonal architecture. This of course requires some mapping,where we mostly useMapStruct for. This necessitates some limitations indata model naming conventions. MapStruct uses a fixed und quite inflexible accessor naming strategy,so you can’t really decide that Protocol Buffers should have one convention but Immutables another.Especially for Immutables we are forced to use JavaBeans-style naming convention, although this isnot a JEE application.