> ## Documentation Index
> Fetch the complete documentation index at: https://docs.flowx.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Creating a Kafka producer

> This guide focuses on creating a **Kafka** producer using Spring Boot.

Here are some tips, including the required configurations and code samples, to help you implement a Kafka producer in Java.

## Required dependencies

Ensure that you have the following dependencies in your project:

```xml theme={"system"}
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.6.1</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.5.1</version>
</dependency>

<dependency>
    <groupId>io.opentracing.contrib</groupId>
    <artifactId>opentracing-kafka-client</artifactId>
    <version>0.1.13</version>
</dependency>
```

## Configuration

Ensure that you have the following configuration in your `application.yml` or `application.properties` file:

```yaml theme={"system"}
spring:
  kafka:
    bootstrap-servers: localhost:9092
    security.protocol: "PLAINTEXT"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        message:
          max:
            bytes: ${KAFKA_MESSAGE_MAX_BYTES:52428800} #50MB
        max:
          request:
            size: ${KAFKA_MESSAGE_MAX_BYTES:52428800} #50MB
```

## Code sample for a Kafka producer

<Warning>
  Ensure that you have the necessary KafkaTemplate bean autowired in your producer class. The sendMessage method demonstrates how to send a message to a Kafka topic with the specified headers and payload. Make sure to include all the received Kafka headers in the response that is sent back to the **FlowX Engine**.
</Warning>

```java theme={"system"}
private final KafkaTemplate<String, Object> kafkaTemplate;

public void sendMessage(String topic, Headers headers, Object payload) {
  ProducerRecord<String, Object> producerRecord = new ProducerRecord<>(topic, payload);
  // make sure to send all the received headers back to the FlowX Engine
  headers.forEach(header -> producerRecord.headers().add(header));
  kafkaTemplate.send(producerRecord);
}
```

## Kafka headers

### Understanding Kafka headers in FlowX Integration

When integrating with FlowX Engine via Kafka, headers play a crucial role in message routing and processing. It's essential to preserve and include all received headers in responses back to the FlowX Engine.

<Info>
  **Need help finding identifier values?** Check out the [Finding Identifiers and Parameters](./finding-identifiers-and-parameters) guide to learn where to locate workspace IDs, application IDs, build IDs, process instance UUIDs, and other identifiers in FlowX Designer.
</Info>

### Important headers

| Header                                    | Purpose                                                              | Consequences If Missing                                                        |
| ----------------------------------------- | -------------------------------------------------------------------- | ------------------------------------------------------------------------------ |
| `fxContext`                               | Identifies the target process/subprocess in the hierarchy            | Messages may be incorrectly routed or not processed                            |
| `Fx-AppId`                                | Identifies the application processing the message                    | Application context may be lost                                                |
| `Fx-RootAppId`                            | Identifies the original application that initiated the process chain | Process origin tracking may be lost; important for complex process hierarchies |
| `Fx-BuildId` (deprecated since 5.0)       | Contains the build identifier for versioning and traceability        | Version-specific behaviors may be affected; debugging becomes more difficult   |
| `Fx-Build-App-Version-Id`                 | Contains the application version identifier for data querying        | Version-specific behaviors may be affected; data queries may fail              |
| `Fx-Workspace-Id`                         | Identifies the workspace context for multi-tenant environments       | Workspace isolation may be compromised; data queries may fail                  |
| `jwt`                                     | JWT token for authenticated operations (process start, task actions) | Operations requiring user context will fail authorization                      |
| `processInstanceId`/`processInstanceUuid` | Primary keys for message correlation                                 | Message correlation may fail                                                   |
| `Fx-ProcessName`                          | Required for cross-process communication                             | "Start another process via Kafka" functionality may break                      |

<Warning>
  **Planning to upgrade to 5.9 LTS?**

  5.9.0 makes `Fx-Organization-Id` mandatory on every Kafka message. Custom Kafka producers that don't send it will be rejected with a non-retryable `MissingKafkaHeaderException`. Review the [Migrating from 5.1 LTS](/5.9/migrating-from-5.1-lts/overview) guide before upgrading.
</Warning>

### The fxContext header explained

The `fxContext` header is particularly important for routing messages in architectures with embedded processes and subprocesses:

* For kafka-receive nodes in the root process: `fxContext = "main"`
* For an embedded subprocess with nodeId=4: `fxContext = "main:4"`
* For an embedded sub-subprocess with nodeId=12: `fxContext = "main:4:12"`

This hierarchical format ensures messages are delivered to their intended recipients within the process structure.

### Understanding Fx-RootAppId

The `Fx-RootAppId` header is used to track the originating application throughout the entire process chain. This is particularly important when:

* Multiple applications are involved in a workflow
* Processes spawn subprocesses across different components
* You need to trace a complete transaction back to its originating application

Unlike `Fx-AppId` which may change as a message passes through different components, `Fx-RootAppId` preserves the original initiator's identity.

### Best Practices

1. **Preserve All Headers**: Always include all received Kafka headers in responses to the FlowX Engine.
   ```java theme={"system"}
   headers.forEach(header -> producerRecord.headers().add(header));
   ```


## Related topics

- [FlowX Admin setup](/5.1/setup-guides/admin-setup-guide.md)
- [Creating a Kafka consumer](/5.1/docs/platform-deep-dive/integrations/creating-a-kafka-consumer.md)
- [Application Manager setup](/5.1/setup-guides/application-manager.md)
- [Building a connector](/5.1/docs/platform-deep-dive/integrations/building-a-connector.md)
- [Finding identifiers and parameters](/5.1/docs/platform-deep-dive/integrations/finding-identifiers-and-parameters.md)