Messaging with Apache Kafka

package com.example.demo.service;

import com.example.demo.dto.UserEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class KafkaProducerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerService.class);
    private static final String USER_EVENTS_TOPIC = "user-events";

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public KafkaProducerService(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendUserEvent(UserEvent event) {
        CompletableFuture<SendResult<String, Object>> future =
            kafkaTemplate.send(USER_EVENTS_TOPIC, event.getUserId().toString(), event);

        future.whenComplete((result, ex) -> {
            if (ex == null) {
                logger.info("Sent message=[{}] with offset=[{}]",
                    event, result.getRecordMetadata().offset());
            } else {
                logger.error("Unable to send message=[{}] due to: {}", event, ex.getMessage());
            }
        });
    }

    public void sendUserCreated(Long userId, String email) {
        UserEvent event = new UserEvent("USER_CREATED", userId, email);
        sendUserEvent(event);
    }

    public void sendUserUpdated(Long userId, String email) {
        UserEvent event = new UserEvent("USER_UPDATED", userId, email);
        sendUserEvent(event);
    }
}

package com.example.demo.service;

import com.example.demo.dto.UserEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerService.class);

    @KafkaListener(topics = "user-events", groupId = "user-service-group")
    public void consumeUserEvent(@Payload UserEvent event,
                                 @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                                 @Header(KafkaHeaders.OFFSET) long offset) {
        logger.info("Received user event: {} from partition: {} with offset: {}",
            event, partition, offset);

        switch (event.getEventType()) {
            case "USER_CREATED":
                handleUserCreated(event);
                break;
            case "USER_UPDATED":
                handleUserUpdated(event);
                break;
            default:
                logger.warn("Unknown event type: {}", event.getEventType());
        }
    }

    private void handleUserCreated(UserEvent event) {
        logger.info("Processing user created: {}", event.getUserId());
        // Send welcome email, create user profile, etc.
    }

    private void handleUserUpdated(UserEvent event) {
        logger.info("Processing user updated: {}", event.getUserId());
        // Invalidate cache, notify other services, etc.
    }
}