Configuring the Message Broker

Apache Kafka is a robust choice for distributed event streaming. Below is a Spring Boot configuration to set up a Kafka producer, specifying the bootstrap servers and serialization strategies.

package com.platform.cashback.setup;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaMessageSetup {

    @Value("${messaging.kafka.servers}")
    private String kafkaServers;

    @Bean
    public ProducerFactory<String, String> kafkaProducerFactory() {
        Map<String, Object> properties = new HashMap<>();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(properties);
    }

    @Bean
    public KafkaTemplate<String, String> eventKafkaTemplate() {
        return new KafkaTemplate<>(kafkaProducerFactory());
    }
}

Publishing Domain Events

When a user makes a purchase that qualifies for a rebate, the system must publish this transaction as an event. The RewardEventPublisher handles dispatching the event data to a designated Kafka topic.

package com.platform.cashback.publisher;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import com.platform.cashback.model.TransactionRecord;

@Component
public class RewardEventPublisher {

    private static final String TOPIC_NAME = "reward_transactions";

    private final KafkaTemplate<String, String> eventKafkaTemplate;

    public RewardEventPublisher(KafkaTemplate<String, String> eventKafkaTemplate) {
        this.eventKafkaTemplate = eventKafkaTemplate;
    }

    public void dispatchTransaction(TransactionRecord record) {
        eventKafkaTemplate.send(TOPIC_NAME, record.getTransactionId(), record.serializePayload());
    }
}

Consuming and Processing Messages

Downstream services must listen to these topics to calculate and allocate the cashback. The RewardEventConsumer utilizes Spring Kafka's listener mechanism to trigger the reward calculation process whenever a new message arrives.

package com.platform.cashback.consumer;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import com.platform.cashback.service.RewardCalculationService;

@Component
public class RewardEventConsumer {

    private final RewardCalculationService rewardCalculationService;

    public RewardEventConsumer(RewardCalculationService rewardCalculationService) {
        this.rewardCalculationService = rewardCalculationService;
    }

    @KafkaListener(topics = "reward_transactions")
    public void receiveTransactionEvent(String payload) {
        rewardCalculationService.computeAndApplyReward(payload);
    }
}