Event-Driven Architecture (Kafka & RabbitMQ)

What is Event-Driven Architecture?

Event-Driven Architecture is a software design pattern where system components communicate through events - notifications of state changes or significant occurrences. Instead of direct service-to-service calls, services produce and consume events asynchronously.

Core Concepts in EDA

1. Event

A record of something that happened in the system. Events are immutable and represent facts.

java

public class OrderCreatedEvent {
    private String eventId;
    private String eventType = "OrderCreated";
    private String orderId;
    private String customerId;
    private BigDecimal amount;
    private LocalDateTime timestamp;
    
    // Constructors, getters, setters
}

2. Event Producer/Publisher

Services that generate events when something significant occurs.

3. Event Consumer/Subscriber

Services that listen for and react to events.

4. Event Broker/Message Broker

Middleware that receives events from producers and routes them to consumers.

Deep Dive: Apache Kafka

Kafka Architecture Overview

Kafka is a distributed, persistent log-based event streaming platform designed for high throughput and fault tolerance.

Core Kafka Components:

1. Topics

A category or feed name to which events are published.

2. Partitions

Topics are split into partitions for parallelism and scalability.

3. Brokers

Kafka servers that store data and serve clients.

4. Producers

Applications that publish events to topics.

5. Consumers

Applications that read events from topics.

6. Consumer Groups

Multiple consumers working together to consume a topic.

Detailed Kafka Code Example

Kafka Configuration

yaml

# application.yml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serialization.JsonSerializer
      acks: all
      retries: 3
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serialization.JsonDeserializer
      group-id: order-service-group
      auto-offset-reset: earliest
      properties:
        spring.json.trusted.packages: "com.example.events"

Event Definitions

java

// Base Event
public abstract class DomainEvent {
    private String eventId;
    private String aggregateId;
    private LocalDateTime timestamp;
    private String eventType;
    private String version = "1.0";
    
    public DomainEvent(String aggregateId) {
        this.eventId = UUID.randomUUID().toString();
        this.aggregateId = aggregateId;
        this.timestamp = LocalDateTime.now();
        this.eventType = this.getClass().getSimpleName();
    }
    // getters
}

// Order Events
public class OrderCreatedEvent extends DomainEvent {
    private String customerId;
    private BigDecimal totalAmount;
    private List<OrderItem> items;
    private String shippingAddress;
    
    public OrderCreatedEvent(String orderId, String customerId, 
                           BigDecimal totalAmount, List<OrderItem> items, 
                           String shippingAddress) {
        super(orderId);
        this.customerId = customerId;
        this.totalAmount = totalAmount;
        this.items = items;
        this.shippingAddress = shippingAddress;
    }
    // getters
}

public class OrderPaidEvent extends DomainEvent {
    private String paymentId;
    private BigDecimal amount;
    private String paymentMethod;
    
    public OrderPaidEvent(String orderId, String paymentId, 
                         BigDecimal amount, String paymentMethod) {
        super(orderId);
        this.paymentId = paymentId;
        this.amount = amount;
        this.paymentMethod = paymentMethod;
    }
    // getters
}

public class OrderShippedEvent extends DomainEvent {
    private String trackingNumber;
    private String carrier;
    private LocalDateTime shippedAt;
    
    public OrderShippedEvent(String orderId, String trackingNumber, 
                           String carrier, LocalDateTime shippedAt) {
        super(orderId);
        this.trackingNumber = trackingNumber;
        this.carrier = carrier;
        this.shippedAt = shippedAt;
    }
    // getters
}

Kafka Producer Service

java

@Service
@Slf4j
public class KafkaEventPublisher {
    
    private final KafkaTemplate<String, Object> kafkaTemplate;
    
    public KafkaEventPublisher(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    public void publishOrderCreated(OrderCreatedEvent event) {
        // Send with orderId as key for partitioning
        ListenableFuture<SendResult<String, Object>> future = 
            kafkaTemplate.send("order-events", event.getAggregateId(), event);
        
        future.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("OrderCreatedEvent published successfully to partition {} with offset {}",
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            }
            
            @Override
            public void onFailure(Throwable ex) {
                log.error("Failed to publish OrderCreatedEvent", ex);
                // Implement retry logic or dead letter queue
            }
        });
    }
    
    public void publishOrderPaid(OrderPaidEvent event) {
        kafkaTemplate.send("order-events", event.getAggregateId(), event)
            .addCallback(
                result -> log.info("OrderPaidEvent published successfully"),
                ex -> log.error("Failed to publish OrderPaidEvent", ex)
            );
    }
    
    // Transactional publishing
    @Transactional
    public void publishOrderEventsInTransaction(Order order) {
        // This ensures database update and event publishing happen atomically
        orderRepository.save(order);
        
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(), order.getCustomerId(), 
            order.getTotalAmount(), order.getItems(), 
            order.getShippingAddress()
        );
        
        publishOrderCreated(event);
    }
}

Kafka Consumer Services

Payment Service Consumer

java

@Service
@Slf4j
public class PaymentServiceConsumer {
    
    private final PaymentService paymentService;
    private final KafkaEventPublisher eventPublisher;
    
    public PaymentServiceConsumer(PaymentService paymentService, 
                                KafkaEventPublisher eventPublisher) {
        this.paymentService = paymentService;
        this.eventPublisher = eventPublisher;
    }
    
    @KafkaListener(
        topics = "order-events",
        containerFactory = "kafkaListenerContainerFactory",
        groupId = "payment-service-group"
    )
    public void handleOrderCreated(OrderCreatedEvent event) {
        log.info("Received OrderCreatedEvent for order: {}", event.getAggregateId());
        
        try {
            // Process payment
            PaymentResult result = paymentService.processPayment(
                event.getAggregateId(),
                event.getTotalAmount(),
                event.getCustomerId()
            );
            
            // Publish payment result
            OrderPaidEvent paidEvent = new OrderPaidEvent(
                event.getAggregateId(),
                result.getPaymentId(),
                result.getAmount(),
                result.getPaymentMethod()
            );
            
            eventPublisher.publishOrderPaid(paidEvent);
            
            log.info("Payment processed successfully for order: {}", event.getAggregateId());
            
        } catch (Exception e) {
            log.error("Failed to process payment for order: {}", event.getAggregateId(), e);
            // In production, you might send to a dead letter topic or retry
        }
    }
    
    // Handle other order-related events if needed
    @KafkaListener(
        topics = "order-events",
        containerFactory = "kafkaListenerContainerFactory",
        groupId = "payment-service-group"
    )
    public void handleOrderCancelled(OrderCancelledEvent event) {
        // Handle refunds when order is cancelled
        paymentService.processRefund(event.getAggregateId());
    }
}

Inventory Service Consumer

java

@Service
@Slf4j
public class InventoryServiceConsumer {
    
    private final InventoryService inventoryService;
    
    public InventoryServiceConsumer(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }
    
    @KafkaListener(
        topics = "order-events",
        containerFactory = "kafkaListenerContainerFactory", 
        groupId = "inventory-service-group"
    )
    public void handleOrderEvents(DomainEvent event) {
        String eventType = event.getClass().getSimpleName();
        String orderId = event.getAggregateId();
        
        switch (eventType) {
            case "OrderCreatedEvent":
                OrderCreatedEvent orderCreated = (OrderCreatedEvent) event;
                inventoryService.reserveItems(orderId, orderCreated.getItems());
                log.info("Inventory reserved for order: {}", orderId);
                break;
                
            case "OrderPaidEvent":
                inventoryService.confirmReservation(orderId);
                log.info("Inventory reservation confirmed for order: {}", orderId);
                break;
                
            case "OrderCancelledEvent":
                inventoryService.releaseReservation(orderId);
                log.info("Inventory reservation released for order: {}", orderId);
                break;
                
            default:
                log.debug("Unhandled event type: {} for order: {}", eventType, orderId);
        }
    }
}

Shipping Service Consumer

java

@Service
@Slf4j
public class ShippingServiceConsumer {
    
    private final ShippingService shippingService;
    private final KafkaEventPublisher eventPublisher;
    
    public ShippingServiceConsumer(ShippingService shippingService,
                                 KafkaEventPublisher eventPublisher) {
        this.shippingService = shippingService;
        this.eventPublisher = eventPublisher;
    }
    
    @KafkaListener(
        topics = "order-events",
        containerFactory = "kafkaListenerContainerFactory",
        groupId = "shipping-service-group"
    )
    public void handleOrderPaid(OrderPaidEvent event) {
        log.info("Order paid, preparing shipment for order: {}", event.getAggregateId());
        
        try {
            ShippingResult result = shippingService.prepareShipment(event.getAggregateId());
            
            // Publish shipping event
            OrderShippedEvent shippedEvent = new OrderShippedEvent(
                event.getAggregateId(),
                result.getTrackingNumber(),
                result.getCarrier(),
                LocalDateTime.now()
            );
            
            eventPublisher.publishOrderShipped(shippedEvent);
            
            log.info("Order shipped with tracking: {}", result.getTrackingNumber());
            
        } catch (Exception e) {
            log.error("Failed to prepare shipment for order: {}", event.getAggregateId(), e);
        }
    }
}

Kafka Configuration Class

java

@Configuration
@EnableKafka
public class KafkaConfig {
    
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        
        return new DefaultKafkaProducerFactory<>(props);
    }
    
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
    
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.events");
        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.events.DomainEvent");
        
        return new DefaultKafkaConsumerFactory<>(props);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3); // Process 3 partitions concurrently
        
        // Configure error handling
        factory.setCommonErrorHandler(errorHandler());
        
        return factory;
    }
    
    @Bean
    public CommonErrorHandler errorHandler() {
        return new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate()),
            new FixedBackOff(1000L, 3) // Retry 3 times with 1-second delay
        );
    }
}

Deep Dive: RabbitMQ

RabbitMQ Architecture Overview

RabbitMQ is a message broker that implements the Advanced Message Queuing Protocol (AMQP). It's designed for reliable message delivery and complex routing.

Core RabbitMQ Components:

1. Exchanges

Receive messages from producers and route them to queues based on rules.

2. Queues

Buffer that stores messages until they are consumed.

3. Bindings

Rules that define how exchanges route messages to queues.

4. Routing Key

Key that determines how messages are routed.

Detailed RabbitMQ Code Example

RabbitMQ Configuration

yaml

# application.yml
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    template:
      exchange: order-events-exchange
    listener:
      simple:
        default-requeue-rejected: false
        acknowledge-mode: auto

RabbitMQ Event Publisher

java

@Service
@Slf4j
public class RabbitMQEventPublisher {
    
    private final AmqpTemplate amqpTemplate;
    private final ObjectMapper objectMapper;
    
    public RabbitMQEventPublisher(AmqpTemplate amqpTemplate, ObjectMapper objectMapper) {
        this.amqpTemplate = amqpTemplate;
        this.objectMapper = objectMapper;
    }
    
    public void publishOrderCreated(OrderCreatedEvent event) {
        try {
            String message = objectMapper.writeValueAsString(event);
            
            amqpTemplate.convertAndSend(
                "order-events-exchange",
                "order.created", // routing key
                message,
                message -> {
                    // Set message properties
                    message.getMessageProperties().setContentType("application/json");
                    message.getMessageProperties().setHeader("event-type", "OrderCreated");
                    message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                    message.getMessageProperties().setTimestamp(new Date());
                    return message;
                }
            );
            
            log.info("OrderCreatedEvent published for order: {}", event.getAggregateId());
            
        } catch (JsonProcessingException e) {
            log.error("Failed to serialize OrderCreatedEvent", e);
            throw new RuntimeException("Event serialization failed", e);
        }
    }
    
    public void publishOrderPaid(OrderPaidEvent event) {
        try {
            String message = objectMapper.writeValueAsString(event);
            
            amqpTemplate.convertAndSend(
                "order-events-exchange", 
                "order.paid", // routing key
                message
            );
            
            log.info("OrderPaidEvent published for order: {}", event.getAggregateId());
            
        } catch (JsonProcessingException e) {
            log.error("Failed to serialize OrderPaidEvent", e);
        }
    }
    
    // Using different exchange types
    public void publishNotificationEvent(NotificationEvent event, String routingKey) {
        // For topic exchange with flexible routing
        amqpTemplate.convertAndSend("notification-exchange", routingKey, event);
    }
    
    public void broadcastSystemEvent(SystemEvent event) {
        // For fanout exchange - broadcast to all queues
        amqpTemplate.convertAndSend("system-events-exchange", "", event);
    }
}

RabbitMQ Consumers

Payment Service Consumer (RabbitMQ)

java

@Service
@Slf4j
public class RabbitMQPaymentServiceConsumer {
    
    private final PaymentService paymentService;
    private final RabbitMQEventPublisher eventPublisher;
    
    public RabbitMQPaymentServiceConsumer(PaymentService paymentService,
                                        RabbitMQEventPublisher eventPublisher) {
        this.paymentService = paymentService;
        this.eventPublisher = eventPublisher;
    }
    
    @RabbitListener(
        queues = "payment-processing-queue",
        ackMode = "AUTO"
    )
    public void handleOrderCreated(OrderCreatedEvent event) {
        log.info("Received OrderCreatedEvent for order: {}", event.getAggregateId());
        
        try {
            PaymentResult result = paymentService.processPayment(
                event.getAggregateId(),
                event.getTotalAmount(),
                event.getCustomerId()
            );
            
            if (result.isSuccess()) {
                OrderPaidEvent paidEvent = new OrderPaidEvent(
                    event.getAggregateId(),
                    result.getPaymentId(),
                    result.getAmount(),
                    result.getPaymentMethod()
                );
                
                eventPublisher.publishOrderPaid(paidEvent);
                log.info("Payment processed successfully for order: {}", event.getAggregateId());
            } else {
                log.error("Payment failed for order: {}", event.getAggregateId());
                // Might send to dead letter exchange or retry queue
            }
            
        } catch (Exception e) {
            log.error("Failed to process payment for order: {}", event.getAggregateId(), e);
            throw new AmqpRejectAndDontRequeueException("Payment processing failed", e);
        }
    }
}

Inventory Service Consumer (RabbitMQ)

java

@Service
@Slf4j
public class RabbitMQInventoryServiceConsumer {
    
    private final InventoryService inventoryService;
    
    public RabbitMQInventoryServiceConsumer(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }
    
    @RabbitListener(queues = "inventory-reservation-queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        log.info("Reserving inventory for order: {}", event.getAggregateId());
        inventoryService.reserveItems(event.getAggregateId(), event.getItems());
    }
    
    @RabbitListener(queues = "inventory-confirmation-queue")
    public void handleOrderPaid(OrderPaidEvent event) {
        log.info("Confirming inventory reservation for order: {}", event.getAggregateId());
        inventoryService.confirmReservation(event.getAggregateId());
    }
    
    @RabbitListener(queues = "inventory-cancellation-queue")
    public void handleOrderCancelled(OrderCancelledEvent event) {
        log.info("Releasing inventory for cancelled order: {}", event.getAggregateId());
        inventoryService.releaseReservation(event.getAggregateId());
    }
}

RabbitMQ Configuration Class

java

@Configuration
public class RabbitMQConfig {
    
    // Exchanges
    @Bean
    public TopicExchange orderEventsExchange() {
        return new TopicExchange("order-events-exchange", true, false);
    }
    
    @Bean
    public FanoutExchange systemEventsExchange() {
        return new FanoutExchange("system-events-exchange", true, false);
    }
    
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("dead-letter-exchange");
    }
    
    // Queues
    @Bean
    public Queue paymentProcessingQueue() {
        return QueueBuilder.durable("payment-processing-queue")
            .withArgument("x-dead-letter-exchange", "dead-letter-exchange")
            .withArgument("x-dead-letter-routing-key", "payment-processing-dlq")
            .build();
    }
    
    @Bean
    public Queue inventoryReservationQueue() {
        return QueueBuilder.durable("inventory-reservation-queue").build();
    }
    
    @Bean
    public Queue inventoryConfirmationQueue() {
        return QueueBuilder.durable("inventory-confirmation-queue").build();
    }
    
    @Bean
    public Queue deadLetterQueue() {
        return new Queue("payment-processing-dlq", true);
    }
    
    // Bindings
    @Bean
    public Binding paymentQueueBinding() {
        return BindingBuilder.bind(paymentProcessingQueue())
               .to(orderEventsExchange())
               .with("order.created");
    }
    
    @Bean
    public Binding inventoryReservationBinding() {
        return BindingBuilder.bind(inventoryReservationQueue())
               .to(orderEventsExchange())
               .with("order.created");
    }
    
    @Bean
    public Binding inventoryConfirmationBinding() {
        return BindingBuilder.bind(inventoryConfirmationQueue())
               .to(orderEventsExchange())
               .with("order.paid");
    }
    
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder.bind(deadLetterQueue())
               .to(deadLetterExchange())
               .with("payment-processing-dlq");
    }
    
    // JSON Message Converter
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    
    @Bean
    public AmqpTemplate amqpTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }
}

Kafka vs RabbitMQ: Detailed Comparison

Architecture Differences

Performance Characteristics

Use Case Scenarios

When to Use Kafka:

java

// Kafka is ideal for:
public class KafkaUseCases {
    // 1. Event Sourcing
    public void eventSourcingExample() {
        // Store all state changes as events
        // Rebuild state by replaying events
    }
    
    // 2. Stream Processing
    public void streamProcessingExample() {
        // Real-time analytics
        // Complex event processing
        // Data pipeline processing
    }
    
    // 3. High-throughput Log Aggregation
    public void logAggregationExample() {
        // Application logs
        // Metrics collection
        // Audit trails
    }
    
    // 4. Replayable Events
    public void replayEventsExample() {
        // Re-process events for new consumers
        // Debugging and testing
        // Data recovery
    }
}

When to Use RabbitMQ:

java

// RabbitMQ is ideal for:
public class RabbitMQUseCases {
    // 1. Task Queues
    public void taskQueueExample() {
        // Background job processing
        // Work distribution
        // Asynchronous task execution
    }
    
    // 2. Complex Routing
    public void complexRoutingExample() {
        // Different consumers based on message content
        // Conditional message delivery
        // Multi-step workflows
    }
    
    // 3. RPC and Request-Reply
    public void rpcExample() {
        // Synchronous communication patterns
        // API gateways
        // Service coordination
    }
    
    // 4. Reliable Delivery
    public void reliableDeliveryExample() {
        // Financial transactions
        // Order processing
        // Critical business operations
    }
}

Message Delivery Guarantees

Kafka Delivery Semantics:

java

public class KafkaDeliveryGuarantees {
    // At-least-once delivery (default)
    Properties props = new Properties();
    props.put("acks", "all"); // Wait for all replicas
    props.put("retries", 3);
    
    // Exactly-once semantics (for critical applications)
    props.put("enable.idempotence", true);
    props.put("transactional.id", "my-transactional-id");
}

// Consumer offset management
@KafkaListener(topics = "my-topic", groupId = "my-group")
public void consumeWithOffsetControl(ConsumerRecord<String, String> record, 
                                   Acknowledgment ack) {
    try {
        processMessage(record.value());
        ack.acknowledge(); // Manual commit after processing
    } catch (Exception e) {
        // Don't acknowledge - message will be redelivered
    }
}

RabbitMQ Delivery Guarantees:

java

public class RabbitMQDeliveryGuarantees {
    // Publisher confirms
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        if (ack) {
            log.info("Message confirmed by broker");
        } else {
            log.error("Message rejected by broker: {}", cause);
        }
    });
    
    // Consumer acknowledgements
    @RabbitListener(queues = "my-queue")
    public void handleMessage(Message message, Channel channel) throws IOException {
        try {
            processMessage(message);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // Negative acknowledgement - requeue or send to DLQ
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }
}

Scalability Patterns

Kafka Scaling:

java

public class KafkaScaling {
    // Horizontal scaling through partitioning
    @Bean
    public NewTopic ordersTopic() {
        return new NewTopic("orders", 6, (short) 3); // 6 partitions, 3 replicas
    }
    
    // Consumer group scaling
    @KafkaListener(topicPartitions = 
        @TopicPartition(topic = "orders", partitions = {"0", "1"})
    )
    public void consumePartitions01(String message) {
        // This consumer handles partitions 0 and 1
    }
}

RabbitMQ Scaling:

java

public class RabbitMQScaling {
    // Vertical scaling with larger instances
    // OR
    // Clustering with mirrored queues
    
    @Bean
    public Queue scalableQueue() {
        return QueueBuilder.durable("scalable-queue")
            .withArgument("x-queue-mode", "lazy") // Disk-based for large queues
            .build();
    }
    
    // Multiple consumers on same queue for load distribution
    @RabbitListener(queues = "scalable-queue", concurrency = "5")
    public void handleMessageConcurrently(String message) {
        // 5 concurrent consumers
    }
}

Summary Table: Kafka vs RabbitMQ

Hybrid Approach

In many modern architectures, both Kafka and RabbitMQ are used together:

java

public class HybridArchitecture {
    // Use Kafka for event sourcing and stream processing
    public void kafkaForEvents() {
        // Order events, user activities, system logs
        // Stream processing for real-time analytics
    }
    
    // Use RabbitMQ for task queues and RPC
    public void rabbitMQForTasks() {
        // Email sending, PDF generation, image processing
        // Request-reply patterns for synchronous operations
    }
}

This extensive guide should give you a solid understanding of Event-Driven Architecture and the differences between Kafka and RabbitMQ. Choose based on your specific requirements around throughput, latency, message retention, and processing patterns.