KafkaPublisherImpl.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.processing.Generated;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Generated(value = "com.asyncapi.generator.template.spring", date = "2025-03-06T09:11:34.123Z")
@Service
public class PublisherServiceImpl implements PublisherService {

    @Autowired
    private KafkaTemplate<Integer, Object> kafkaTemplate;

    // Example method for publishing messages
    public void publishMessage(Integer key, YourMessageType messagePayload) {
        Message<YourMessageType> message = MessageBuilder.withPayload(messagePayload)
                .setHeader(KafkaHeaders.TOPIC, "yourTopicName")
                .setHeader(KafkaHeaders.KEY, key)
                .build();

        // Asynchronous send with callback
        ListenableFuture<SendResult<Integer, Object>> future = kafkaTemplate.send(message);
        future.addCallback(new ListenableFutureCallback<SendResult<Integer, Object>>() {
            @Override
            public void onSuccess(SendResult<Integer, Object> result) {
                // Handle success scenario
                System.out.println("Message sent successfully: " + result.getProducerRecord().value());
            }

            @Override
            public void onFailure(Throwable ex) {
                // Handle failure scenario
                System.err.println("Error sending message: " + ex.getMessage());
                // Implement retry logic or other error handling mechanisms as needed
            }
        });

        // Alternatively, for synchronous send with exception handling
        try {
            kafkaTemplate.send(message).get(10, TimeUnit.SECONDS);
            // Handle success scenario
            System.out.println("Message sent successfully: " + message.getPayload());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            // Handle thread interruption
            System.err.println("Thread interrupted while sending message: " + e.getMessage());
        } catch (ExecutionException e) {
            // Handle execution exceptions
            System.err.println("Execution exception during message send: " + e.getMessage());
            // Implement retry logic or other error handling mechanisms as needed
        } catch (TimeoutException e) {
            // Handle timeout exceptions
            System.err.println("Timeout while sending message: " + e.getMessage());
            // Implement retry logic or other error handling mechanisms as needed
        }
    }

}