EventPublisherService.java

@Service
public class EventPublisherService {
    @Autowired
    private OutboxRepository outboxRepository;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Scheduled(fixedRate = 10000) // Every 10 seconds
    public void publishEvents() {
        List<Outbox> events = outboxRepository.findAll();
        for (Outbox event : events) {
            kafkaTemplate.send("messages-topic", event.getPayload()).thenAccept(sendResult -> {
                outboxRepository.delete(event);
            }).exceptionally(throwable -> {
                // Log error or implement retry mechanism
                return null;
            });
        }
    }
}