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 60 seconds.

Note that the second off reading produced an intermediate result.

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

These are deliberate choices and can easily be changed.

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

The data model is in the model directory, the business logic in topology.

The tests test the topology with seven different (de-)serializers, protocol buffers, JSON, Apache Avro, the Confluent variants of them and Amazon Ion. Since the example needs an input format, a result format and a format for the state store we have 343 (7³) different combinations which are all tested.

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

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

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

Since the business logic is independent of the serialization mechanism, changing it is simple and normally does not require retesting. Be aware of subtle pitfalls tough, while our Java code uses nanosecond precision, which are correctly handled with protocol buffers and JSON, Avro truncates the time stamps to microseconds. We would not catch these in our tests since we use only second steps for testing, but this would show up when using real world data and comparing with equals.

By refactoring the business logic to depend only on an abstract store, we speed up testing by a factor of twenty (bazel test //src/test/java/de/melsicon/kafka/sensors/logic:all vs. //src/test/java/de/melsicon/kafka/sensors/topology:all), which demonstrates a potential for improvement in development speed and testability.

You need Bazelisk installed, with HomeBrew just use brew install bazelbuild/tap/bazelisk.

To run all tests, use

To run a single test, use

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

The main app needs Kafka running at localhost, port 9092 (see application.yaml). Start it with

Watch the results with

and publish sensor values with

The main app runs with JSON in- and output, so you can use sample sensor values like

Run the JMH microbenchmarks with

As noted in Implementation of Business Logic the business login is independent of the serialization, in the spirit of hexagonal architecture. This of course requires some mapping, where we mostly use MapStruct for. This necessitates some limitations in data model naming conventions. MapStruct uses a fixed und quite unflexible 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 is not a JEE application.