Event-Driven Architecture

@Service
@RequiredArgsConstructor
public class PatientEventPublisher {
    
    private final KafkaTemplate kafkaTemplate;
    private final EventStore eventStore;
    
    @Transactional
    public void enrollPatient(EnrollPatientCommand command) {
        // Create domain event
        PatientEnrolledEvent event = PatientEnrolledEvent.builder()
            .patientId(command.getPatientId())
            .studyId(command.getStudyId())
            .siteId(command.getSiteId())
            .enrolledAt(Instant.now())
            .enrolledBy(command.getEnrolledBy())
            .build();
        
        // Store in event log (source of truth)
        eventStore.append("patient", command.getPatientId(), event);
        
        // Publish to Kafka for downstream consumers
        kafkaTemplate.send(
            "patient.enrolled",
            command.getPatientId(),  // Partition key for ordering
            event
        ).addCallback(
            result -> log.info("Event published: {}", event),
            ex -> log.error("Failed to publish event", ex)
        );
    }
}

@Component
@RequiredArgsConstructor
public class NotificationConsumer {
    
    private final NotificationService notificationService;
    
    @KafkaListener(
        topics = "patient.enrolled",
        groupId = "notification-service",
        containerFactory = "kafkaListenerContainerFactory"
    )
    @Retryable(
        value = {TransientException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void onPatientEnrolled(
        @Payload PatientEnrolledEvent event,
        @Header(KafkaHeaders.RECEIVED_KEY) String patientId,
        Acknowledgment ack
    ) {
        try {
            // Send welcome notification
            notificationService.sendWelcomeEmail(event);
            notificationService.sendSMSConfirmation(event);
            
            // Commit offset only after successful processing
            ack.acknowledge();
            
        } catch (TransientException e) {
            // Will be retried automatically
            throw e;
        } catch (Exception e) {
            // Non-retryable: send to DLQ
            log.error("Failed to process event: {}", event, e);
            throw new NonRetryableException(e);
        }
    }
}