Kafka

Sends events to specified Kafka topics, enabling real-time data streams

Overview

The Kafka Publisher Transport enables publishing of processed events to a specified Kafka topic on a defined Kafka cluster. Events are serialised using either a default JSON serialiser (StreamEventJsonSerializer) or a custom-defined transformer.

Configuration includes setting the Kafka cluster address in the /etc/hosts file and defining topics and partition keys for message distribution. Key configuration attributes are prioritised by importance and cover serialisation options, allowing flexible data handling.

Out-of-the-box serialisers support common formats (JSON, Avro), while custom serialisers and transformers facilitate integration with downstream applications by converting events into domain-specific data types.

The Kafka Publisher setup supports scalable, resilient event processing workflows, suitable for chaining multiple data-processing stages.

The Kafka consumers relies on the client library org.apache.kafka:kafka-clients:2.7.0

Examples & DSL attributes

In this example, the Kafka Publisher sends events to the customers topic on the Kafka broker (KAFKA_BROKER:9092).

Events are partitioned by customerId, ensuring that all events for the same customer go to the same partition for ordered processing.

Using default settings, events are serialized as JSON StreamEvent objects, making them easily readable for downstream processes. This setup enables chaining multiple processing stages, useful for building scalable and resilient data workflows.

kafkaPublisher:
  cluster address: KAFKA_BROKER:9092
  topic: customers
  partitionKeys:
    - customerId

Add to the /etc/hosts file the address of the Kafka host i.e. 127.0.0.1 KAFKA_BROKER

Attributes schema

Configuration parameters available for the InfluxDB publisher transport.

The parameters are organised by order of importance, ranked from high to low.

Attribute	Description	Data Type	Required
cluster address	InfluxDB server address	http://<ip-address>:port
topic	InfluxDB UI / CLI Authentication access token	String
partitionKeys	Authentication access details for v1.x InfluxDB. Note: Only required if authToken has not been provided	String
serializer	InfluxDB UI / CLI organisation token	String
batchSize	Number of events to batch send, maps to batch.size. Batch sise of zero disables batching function	Integer Default: 1024
memBufferSize	Size in bytes of event buffer. Maps to buffer.memory	Long Default: 33554432
retries	Maps to retries	Integer Default: 0
properties	Additional publisher properties to be applied to the Kafka publisher subsystem	Properties map

Serialisation attributes schema

Attribute	Description	Data Type	Required
transform	User provided implementation	Implementation of CustomTransformer
key serializer	Domain class that maps to the partition key type. Property maps to key.serializer property	String Default: IntegerSerializer
value serializer	Domain class that serializes to Kafka. Property maps to value.serializer property	String Default: StreamEventJsonSerializer

Serialiser example

This example configures the Kafka Publisher to send events to the customers topic on a local broker (localhost:9092). Events are partitioned by customerId, ensuring ordered processing per customer.

1. key serializer IntegerSerializer for customerId.

2. value serializer StreamEventJsonSerializer for JSON-formatted event data.

This setup organises events by "customer ID" and makes them easily readable for downstream processing.

kafkaPublisher:
  cluster address: localhost:9092
  topic: customers
  partitionKeys:
    - customerId

  serializer:
    key serializer: org.apache.kafka.common.serialization.IntegerSerializer
    value serializer: com.fractalworks.streams.transport.kafka.serializers.json.StreamEventJsonSerializer

OOTB serialisers

A flexible event serialisation model allows Joule processes to be chained together for complex use cases and easy downstream integration.

Joule includes built-in serialisers, compatible with Apache Kafka, and also supports custom serialisation through the Joule SDK.

com.fractalworks.streams.transport.kafka.serializers.object.ObjectSerializer
com.fractalworks.streams.transport.kafka.serializers.json.StreamEventJsonSerializer
com.fractalworks.streams.transport.kafka.serializers.avro.AvroStreamEventSerializer

Custom transformers

By implementing a custom transformer, you can create domain-specific data types to simplify integration with downstream applications.

Example

In this example, the Kafka Publisher sends events to the customers topic on a local Kafka broker (localhost:9092).

Events are partitioned by customerId and serialised using custom serialisers:

1. key serializer IntegerSerializer for the customerId (integer).

2. value serializer ObjectSerializer for the event data.

3. transformer The CustomerTransformer is used to convert event data into a domain-specific Customer object before serialisation.

This setup enables custom data transformations and ensures events are processed and stored efficiently.

kafkaPublisher:
  cluster address: localhost:9092
  topic: customers
  partitionKeys:
    - customerId
 
  serializer:
    transform: com.fractalworks.streams.transport.kafka.CustomerTransformer
    key serializer: org.apache.kafka.common.serialization.IntegerSerializer
    value serializer: com.fractalworks.streams.transport.kafka.serializers.object.ObjectSerializer

Java transformer implementation

This class will convert an internal StreamEvent object to a user defined Customer data type.

public class CustomerTransformer implements CustomTransformer<Customer> {

   public CustomerTransformer() {
       // Required
   }

   @Override
   public Collection<Customer> transform(Collection<StreamEvent> events) 
   throws TranslationException {

       Collection<Customer> customers = new ArrayList<>();
       for (StreamEvent event : events) {
           customers.add(transform(event));
       }
       return customers;
   }

   @Override
   public Customer transform(StreamEvent event) 
   throws TranslationException {

       Customer customer = new Customer();
       customer.setCustomerId( (int)event.getValue("customerId"));
       customer.setFirstname( (String) event.getValue("firstname"));
       customer.setSurname( (String) event.getValue("surname"));
       return customer;
   }
}