Inter-Service Communication

@Service
@RequiredArgsConstructor
public class OrderService {

    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;

    @Transactional
    public String placeOrder(OrderRequest request) {
        Order order = orderRepository.save(mapToOrder(request));

        // DANGER: What if this send() fails AFTER the DB commits?
        // Result: Order exists in DB but no event published — downstream services never know
        //
        // DANGER: What if the DB transaction rolls back AFTER Kafka accepted the message?
        // Result: Event published for an order that doesn't exist
        kafkaTemplate.send("order-events", order.getOrderNumber(),
            new OrderPlacedEvent(order.getOrderNumber(), order.getItems()));

        return order.getOrderNumber();
    }
}

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxRepository outboxRepository;
    private final ObjectMapper objectMapper;

    @Transactional  // Single atomic transaction — BOTH succeed or BOTH fail
    public String placeOrder(OrderRequest request) {
        Order order = orderRepository.save(mapToOrder(request));

        // Write event to outbox table in the SAME transaction
        OutboxEvent event = OutboxEvent.builder()
            .id(UUID.randomUUID())
            .aggregateId(order.getOrderNumber())
            .aggregateType("Order")
            .eventType("OrderPlaced")
            .payload(objectMapper.writeValueAsString(
                new OrderPlacedEvent(order.getOrderNumber(), order.getItems())))
            .createdAt(Instant.now())
            .build();
        outboxRepository.save(event);

        return order.getOrderNumber();
        // Debezium CDC picks up the outbox row from the WAL and publishes to Kafka
    }
}

@KafkaListener(topics = "order-events", groupId = "notification-group")
public void handleOrderPlaced(OrderPlacedEvent event, Acknowledgment ack) {
    ack.acknowledge();  // Commit offset FIRST
    // If processing fails after this point, message is lost forever
    emailService.sendOrderConfirmation(event);
}

@KafkaListener(topics = "order-events", groupId = "inventory-group")
public void handleOrderPlaced(OrderPlacedEvent event, Acknowledgment ack) {
    // Process first
    inventoryService.reserveStock(event.getOrderNumber(), event.getItems());
    ack.acknowledge();  // Commit offset AFTER successful processing
    // If crash happens before ack, message will be redelivered (processed TWICE)
}

@Service
@RequiredArgsConstructor
public class IdempotentOrderConsumer {

    private final ProcessedEventRepository processedEventRepo;
    private final InventoryService inventoryService;

    @KafkaListener(topics = "order-events", groupId = "inventory-group")
    @Transactional
    public void handleOrderPlaced(OrderPlacedEvent event, Acknowledgment ack) {
        // Idempotency check: have we processed this event before?
        if (processedEventRepo.existsByEventId(event.getEventId())) {
            log.info("Duplicate event {} — skipping", event.getEventId());
            ack.acknowledge();
            return;
        }

        // Process the event
        inventoryService.reserveStock(event.getOrderNumber(), event.getItems());

        // Record that we processed it (in SAME transaction as the business logic)
        processedEventRepo.save(new ProcessedEvent(event.getEventId(), Instant.now()));

        ack.acknowledge();
    }
}

@Service
public class OrderEventProducer {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void publishOrderEvent(OrderEvent event) {
        // Use orderId as key — all events for same order go to same partition
        // This guarantees: OrderCreated → PaymentProcessed → Shipped arrive in order
        kafkaTemplate.send("order-events", event.getOrderId(), event);
    }

    // WRONG: Using random/null key — messages scatter across partitions
    public void publishWrong(OrderEvent event) {
        kafkaTemplate.send("order-events", event);  // null key = round-robin partition
        // OrderCreated might arrive AFTER PaymentProcessed!
    }
}

@Component
@Slf4j
public class ConsumerLagMonitor {

    private final AdminClient adminClient;
    private final MeterRegistry meterRegistry;

    @Scheduled(fixedDelay = 10_000)
    public void checkConsumerLag() {
        Map<TopicPartition, OffsetAndMetadata> committed = adminClient
            .listConsumerGroupOffsets("order-group")
            .partitionsToOffsetAndMetadata().get();

        Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets = adminClient
            .listOffsets(committed.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest())))
            .all().get();

        for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : committed.entrySet()) {
            long lag = endOffsets.get(entry.getKey()).offset() - entry.getValue().offset();

            meterRegistry.gauge("kafka.consumer.lag",
                Tags.of("topic", entry.getKey().topic(),
                         "partition", String.valueOf(entry.getKey().partition())),
                lag);

            if (lag > 10000) {
                log.error("CRITICAL: Consumer lag {} on {}", lag, entry.getKey());
            }
        }
    }
}

@Configuration
public class BackpressureConfig {

    @Bean
    public ConsumerFactory<String, OrderEvent> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");

        // BACKPRESSURE: Only fetch 50 records per poll
        // If processing is slow, fewer records means less memory pressure
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);

        // If processing takes longer than this, consumer is considered dead
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300_000); // 5 minutes

        // Fetch at most 1MB per partition per poll
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1_048_576);

        return new DefaultKafkaConsumerFactory<>(props);
    }
}

@Component
@Slf4j
public class AdaptiveConsumer implements ConsumerSeekAware {

    private final KafkaListenerEndpointRegistry registry;
    private final AtomicInteger processingQueue = new AtomicInteger(0);

    private static final int HIGH_WATERMARK = 1000;
    private static final int LOW_WATERMARK = 200;

    @KafkaListener(id = "orderConsumer", topics = "order-events")
    public void consume(OrderPlacedEvent event) {
        processingQueue.incrementAndGet();
        try {
            processOrder(event);
        } finally {
            int remaining = processingQueue.decrementAndGet();
            if (remaining <= LOW_WATERMARK) {
                resumeConsumer();
            }
        }
    }

    @Scheduled(fixedDelay = 1000)
    public void checkBackpressure() {
        if (processingQueue.get() >= HIGH_WATERMARK) {
            log.warn("Backpressure: pausing consumer (queue size: {})", processingQueue.get());
            pauseConsumer();
        }
    }

    private void pauseConsumer() {
        MessageListenerContainer container = registry.getListenerContainer("orderConsumer");
        if (container != null && container.isRunning()) {
            container.pause();
        }
    }

    private void resumeConsumer() {
        MessageListenerContainer container = registry.getListenerContainer("orderConsumer");
        if (container != null && container.isPauseRequested()) {
            container.resume();
        }
    }
}

@Component
public class TracingProducerInterceptor implements ProducerInterceptor<String, Object> {

    @Override
    public ProducerRecord<String, Object> onSend(ProducerRecord<String, Object> record) {
        // Inject current trace context into Kafka headers
        Span currentSpan = Span.current();
        if (currentSpan != null) {
            record.headers().add("X-Trace-Id", 
                currentSpan.getSpanContext().getTraceId().getBytes());
            record.headers().add("X-Span-Id",
                currentSpan.getSpanContext().getSpanId().getBytes());
            record.headers().add("X-Correlation-Id",
                MDC.get("correlationId").getBytes());
        }
        return record;
    }

    // ... other methods
}

@Component
public class TracingConsumerInterceptor {

    @KafkaListener(topics = "order-events", groupId = "inventory-group")
    public void handleOrderPlaced(
            @Payload OrderPlacedEvent event,
            @Header("X-Trace-Id") byte[] traceId,
            @Header("X-Correlation-Id") byte[] correlationId) {

        // Restore trace context for downstream tracing
        MDC.put("traceId", new String(traceId));
        MDC.put("correlationId", new String(correlationId));

        log.info("Processing order {} [traceId={}, correlationId={}]",
            event.getOrderNumber(), new String(traceId), new String(correlationId));

        try {
            inventoryService.reserveStock(event);
        } finally {
            MDC.clear();
        }
    }
}

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> kafkaListenerContainerFactory(
            ConsumerFactory<String, OrderEvent> consumerFactory,
            KafkaTemplate<String, OrderEvent> kafkaTemplate) {

        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // Retry 3 times, then send to DLQ
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, ex) -> new TopicPartition(record.topic() + ".DLQ", record.partition())),
            new FixedBackOff(1000L, 3L)  // 1 second delay, 3 retries
        );

        // Don't retry on deserialization errors (definitely poison)
        errorHandler.addNotRetryableExceptions(
            DeserializationException.class,
            ClassCastException.class,
            NullPointerException.class
        );

        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}

@Service
@Slf4j
public class DlqAlertConsumer {

    private final AlertService alertService;
    private final MeterRegistry meterRegistry;

    @KafkaListener(topics = "order-events.DLQ", groupId = "dlq-monitor")
    public void handleDeadLetter(
            ConsumerRecord<String, byte[]> record,
            @Header(KafkaHeaders.EXCEPTION_MESSAGE) String errorMessage,
            @Header(KafkaHeaders.ORIGINAL_TOPIC) String originalTopic) {

        log.error("Dead letter received from topic {}: key={}, error={}",
            originalTopic, record.key(), errorMessage);

        // Increment DLQ counter for alerting
        meterRegistry.counter("kafka.dlq.messages",
            "original_topic", originalTopic).increment();

        // Alert if DLQ rate exceeds threshold
        alertService.alertIfThresholdExceeded("kafka.dlq.messages", 10, Duration.ofMinutes(5));
    }
}

@Configuration
public class KafkaAvroConfig {

    @Bean
    public ProducerFactory<String, OrderPlacedEvent> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        props.put("schema.registry.url", "http://schema-registry:8081");
        // Auto-register schema (disable in production — use CI/CD)
        props.put("auto.register.schemas", false);
        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    public ConsumerFactory<String, OrderPlacedEvent> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        props.put("schema.registry.url", "http://schema-registry:8081");
        props.put("specific.avro.reader", true);
        return new DefaultKafkaConsumerFactory<>(props);
    }
}