Ensure that you have the following configuration in your application.yml or application.properties file:
Copy
Ask AI
spring.kafka: bootstrap-servers: URL_OF_THE_KAFKA_SERVER producer: key-deserializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer properties: interceptor: classes: io.opentracing.contrib.kafka.TracingProducerInterceptor security.protocol: "SASL_PLAINTEXT" sasl.mechanism: "OAUTHBEARER" sasl.jaas.config: "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ;" sasl.login.callback.handler.class: io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandlerkafka: authorizationExceptionRetryInterval: 10 ADD_NEEDED_TOPIC_NAMES_HERE # make sure to use the correct naming pattern for topics used to send data to the FLOWX Engine
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.
Copy
Ask AI
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);}