Creating a Kafka producer

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:

<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:

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

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.

​

Important headers

​

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.

   headers.forEach(header -> producerRecord.headers().add(header));