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.

Serialisation attributes schema

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;
   }
}

Default producer properties

acks=0
retries=1
linger.ms=5
buffer.memory=33554432
request.timeout.ms=60000
key.serializer=org.apache.kafka.common.serialization.IntegerSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
partitioner.class=com.fractalworks.streams.transport.kafka.KeyPartitioner

Last updated