# Kafka ## 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**. {% hint style="info" %} The Kafka consumers relies on the client library [org.apache.kafka:kafka-clients:2.7.0](https://kafka.apache.org/documentation/) {% endhint %} ## 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**. ```yaml kafkaPublisher: cluster address: KAFKA_BROKER:9092 topic: customers partitionKeys: - customerId ``` {% hint style="danger" %} Add to the `/etc/hosts` file the address of the Kafka host i.e. 127.0.0.1 KAFKA\_BROKER {% endhint %} ### 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	true
topic	InfluxDB UI / CLI Authentication access token	String	true
partitionKeys	Authentication access details for v1.x InfluxDB. Note: Only required if authToken has not been provided	String	true
serializer	InfluxDB UI / CLI organisation token	String	true
batchSize	Number of events to batch send, maps to batch.size. Batch sise of zero disables batching function	Integer Default: 1024	false
memBufferSize	Size in bytes of event buffer. Maps to `buffer.memory`	Long Default: 33554432	false
retries	Maps to retries	Integer Default: 0	false
properties	Additional publisher properties to be applied to the Kafka publisher subsystem. Default properties applied are listed below.	Properties map	false

### **Serialisation** attributes schema

Attribute Description Data Type Required

transform User provided implementation Implementation of CustomTransformer false

key serializer

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

Domain class that maps to the partition key type. Property maps to key.serializer property

String

Default: IntegerSerializer

false

value serializer

Domain class that serializes to Kafka. Property maps to value.serializer property

String

Default: StreamEventJsonSerializer

false

### **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. ```yaml 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](/joule/developer-guides/builder-sdk/connector-api.md). ```yaml 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. ```yaml 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. ```java public class CustomerTransformer implements CustomTransformer { public CustomerTransformer() { // Required } @Override public Collection transform(Collection events) throws TranslationException { Collection 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 ```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 ``` --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.fractalworks.io/joule/components/connectors/sinks/kafka.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.