Event-Driven архитектура с Kafka: Outbox Pattern, Exactly-Once и идемпотентные консьюмеры

@Transactional
public Order createOrder(OrderRequest request) {
    Order order = new Order(request);
    orderRepository.save(order);
    
    OutboxEvent event = OutboxEvent.builder()
        .aggregateType("Order")
        .aggregateId(order.getId())
        .eventType("OrderCreated")
        .payload(objectMapper.writeValueAsString(
            new OrderCreatedEvent(order)))
        .build();
    
    outboxRepository.save(event);
    
    return order;
}

@Scheduled(fixedDelay = 1000)
public void publishEvents() {
    List<OutboxEvent> events = outboxRepository
        .findTop100ByPublishedFalseOrderByCreatedAt();
    
    for (OutboxEvent event : events) {
        try {
            kafkaTemplate.send(
                getTopicName(event.getEventType()),
                event.getAggregateId(),
                event.getPayload()
            ).get(5, TimeUnit.SECONDS);
            
            event.setPublished(true);
            event.setProcessedAt(Instant.now());
            outboxRepository.save(event);
        } catch (Exception e) {
            log.error("Failed to publish event {}", event.getId(), e);
            // Retry на следующей итерации
        }
    }
}

@Entity
@Table(name = "outbox_events", indexes = {
    @Index(name = "idx_published", columnList = "published,createdAt")
})
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(nullable = false)
    private String aggregateType;  // Order, Payment, User
    
    @Column(nullable = false)
    private String aggregateId;     // UUID сущности
    
    @Column(nullable = false)
    private String eventType;       // OrderCreated, PaymentProcessed
    
    @Column(columnDefinition = "jsonb", nullable = false)
    private String payload;
    
    @Column(nullable = false)
    private Instant createdAt = Instant.now();
    
    private Instant publishedAt;
    
    @Column(nullable = false)
    private boolean published = false;
    
    // getters, setters, builder
}

@Repository
public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long> {
    
    // Берем топ-N необработанных, сортируя по времени создания
    @Query("SELECT e FROM OutboxEvent e WHERE e.published = false " +
           "ORDER BY e.createdAt ASC")
    List<OutboxEvent> findUnpublishedEvents(Pageable pageable);
    
    // Для cleanup старых записей
    @Modifying
    @Query("DELETE FROM OutboxEvent e WHERE e.published = true " +
           "AND e.publishedAt < :threshold")
    void deleteOldPublishedEvents(@Param("threshold") Instant threshold);
}

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;
    
    @Transactional
    public Order createOrder(CreateOrderRequest request) {
        // Основная бизнес-логика
        Order order = Order.builder()
            .userId(request.getUserId())
            .items(request.getItems())
            .totalAmount(calculateTotal(request.getItems()))
            .status(OrderStatus.CREATED)
            .build();
        
        orderRepository.save(order);
        
        // Создаём событие для outbox
        OrderCreatedEvent eventPayload = new OrderCreatedEvent(
            order.getId(),
            order.getUserId(),
            order.getTotalAmount(),
            order.getItems()
        );
        
        OutboxEvent outboxEvent = OutboxEvent.builder()
            .aggregateType("Order")
            .aggregateId(order.getId().toString())
            .eventType("OrderCreated")
            .payload(objectMapper.writeValueAsString(eventPayload))
            .build();
        
        outboxRepository.save(outboxEvent);
        
        return order;
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class OutboxEventPublisher {
    private final OutboxEventRepository repository;
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    @Scheduled(fixedDelay = 1000) // каждую секунду
    @Transactional
    public void publishPendingEvents() {
        List<OutboxEvent> events = repository.findUnpublishedEvents(
            PageRequest.of(0, 100)
        );
        
        if (events.isEmpty()) {
            return;
        }
        
        log.info("Publishing {} events", events.size());
        
        for (OutboxEvent event : events) {
            try {
                // Синхронная отправка с таймаутом
                SendResult<String, String> result = kafkaTemplate.send(
                    resolveTopicName(event.getEventType()),
                    event.getAggregateId(),
                    event.getPayload()
                ).get(5, TimeUnit.SECONDS);
                
                // Помечаем только после успешной отправки
                event.setPublished(true);
                event.setPublishedAt(Instant.now());
                repository.save(event);
                
                log.debug("Published event {} to offset {}", 
                    event.getId(), 
                    result.getRecordMetadata().offset());
                    
            } catch (Exception e) {
                log.error("Failed to publish event {}, will retry", 
                    event.getId(), e);
                // Оставляем published=false, retry на следующей итерации
            }
        }
    }
    
    private String resolveTopicName(String eventType) {
        // Маппинг типа события на топик Kafka
        return switch (eventType) {
            case "OrderCreated", "OrderCancelled" -> "orders";
            case "PaymentProcessed" -> "payments";
            default -> "default-events";
        };
    }
}

@Scheduled(cron = "0 0 2 * * *") // каждую ночь в 2:00
@Transactional
public void cleanupOldEvents() {
    Instant threshold = Instant.now().minus(7, ChronoUnit.DAYS);
    int deleted = repository.deleteOldPublishedEvents(threshold);
    log.info("Cleaned up {} old outbox events", deleted);
}

@Configuration
public class KafkaProducerConfig {
    
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
            StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
            StringSerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        
        return new DefaultKafkaProducerFactory<>(props);
    }
}

@Repository
public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long> {
 
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @QueryHints({
        @QueryHint(name = "javax.persistence.lock.timeout", value = "0")
    })
    @Query(value = "SELECT e FROM OutboxEvent e WHERE e.published = false " +
                   "ORDER BY e.createdAt ASC")
    List<OutboxEvent> findUnpublishedEventsWithLock(Pageable pageable);
}

@Query(value = "SELECT * FROM outbox_events " +
               "WHERE published = false " +
               "ORDER BY created_at " +
               "LIMIT :batchSize " +
               "FOR UPDATE SKIP LOCKED",
       nativeQuery = true)
List<OutboxEvent> lockAndFetchEvents(@Param("batchSize") int batchSize);

@Value("${outbox.publisher.instance-id:0}")
private int publisherInstanceId;
 
@Value("${outbox.publisher.total-instances:1}")
private int totalPublisherInstances;
 
@Scheduled(fixedDelay = 1000)
@Transactional
public void publishPendingEvents() {
    List<OutboxEvent> events = repository.findUnpublishedEventsByPartition(
        publisherInstanceId,
        totalPublisherInstances,
        PageRequest.of(0, 100)
    );
    // processing как раньше
}

@Query("SELECT e FROM OutboxEvent e WHERE e.published = false " +
       "AND MOD(e.id, :totalInstances) = :instanceId " +
       "ORDER BY e.createdAt ASC")
List<OutboxEvent> findUnpublishedEventsByPartition(
    @Param("instanceId") int instanceId,
    @Param("totalInstances") int totalInstances,
    Pageable pageable
);

@Component
@RequiredArgsConstructor
public class OutboxMetrics {
    private final OutboxEventRepository repository;
    private final MeterRegistry meterRegistry;
 
    @Scheduled(fixedDelay = 10000) // каждые 10 секунд
    public void recordMetrics() {
        long unpublishedCount = repository.countByPublishedFalse();
        meterRegistry.gauge("outbox.queue.size", unpublishedCount);
        
        // Максимальный возраст необработанного события
        repository.findTopByPublishedFalseOrderByCreatedAtAsc()
            .ifPresent(oldest -> {
                long ageSeconds = Duration.between(
                    oldest.getCreatedAt(), 
                    Instant.now()
                ).getSeconds();
                meterRegistry.gauge("outbox.max.age.seconds", ageSeconds);
            });
    }
}

@SpringBootTest
@Transactional
class OutboxPatternIntegrationTest {
 
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private OutboxEventRepository outboxRepository;
    
    @Autowired
    private OutboxEventPublisher publisher;
 
    @Test
    @Rollback(false)
    void shouldPublishEventAfterOrderCreation() {
        // Given
        CreateOrderRequest request = new CreateOrderRequest(
            "user-123", 
            List.of(new OrderItem("item-1", 2))
        );
        
        // When
        Order order = orderService.createOrder(request);
        
        // Then - событие в outbox
        List<OutboxEvent> events = outboxRepository.findAll();
        assertThat(events).hasSize(1);
        OutboxEvent event = events.get(0);
        assertThat(event.getEventType()).isEqualTo("OrderCreated");
        assertThat(event.getAggregateId()).isEqualTo(order.getId().toString());
        assertThat(event.isPublished()).isFalse();
        
        // When - publisher обрабатывает
        publisher.publishPendingEvents();
        
        // Then - событие помечено как опубликованное
        OutboxEvent published = outboxRepository.findById(event.getId()).get();
        assertThat(published.isPublished()).isTrue();
        assertThat(published.getPublishedAt()).isNotNull();
        
        // Cleanup
        outboxRepository.deleteAll();
    }
}

@Test
void shouldRetryFailedPublishing() throws Exception {
    // Создаем событие
    orderService.createOrder(request);
    
    // Останавливаем Kafka
    kafkaContainer.stop();
    
    // Попытка публикации должна зафейлиться
    publisher.publishPendingEvents();
    
    // Событие всё еще unpublished
    assertThat(outboxRepository.countByPublishedFalse()).isEqualTo(1);
    
    // Запускаем Kafka обратно
    kafkaContainer.start();
    
    // Retry должен сработать
    publisher.publishPendingEvents();
    
    // Теперь опубликовано
    assertThat(outboxRepository.countByPublishedFalse()).isZero();
}

@Component
public class OutboxBatchWriter {
    private static final int BATCH_SIZE = 50;
    private final ThreadLocal<List<OutboxEvent>> buffer = 
        ThreadLocal.withInitial(ArrayList::new);
        
    @Autowired
    private OutboxEventRepository repository;
 
    public void add(OutboxEvent event) {
        List<OutboxEvent> events = buffer.get();
        events.add(event);
        
        if (events.size() >= BATCH_SIZE) {
            flush();
        }
    }
    
    public void flush() {
        List<OutboxEvent> events = buffer.get();
        if (!events.isEmpty()) {
            repository.saveAll(events);
            events.clear();
        }
    }
}

wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

producer.initTransactions();
 
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  
  producer.beginTransaction();
  try {
      for (ConsumerRecord<String, String> record : records) {
          // Обработка и отправка результата
          String result = process(record.value());
          producer.send(new ProducerRecord<>("output", record.key(), result));
      }
      
      // Коммит offset в той же транзакции
      producer.sendOffsetsToTransaction(
          getCurrentOffsets(records),
          consumer.groupMetadata()
      );
      
      producer.commitTransaction();
  } catch (Exception e) {
      producer.abortTransaction();
      // Retry со следующего poll, offset не сдвинулся
  }
}

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
 
// Обязательные параметры для exactly-once
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-producer-1");
props.put(ProducerConfig.ACKS_CONFIG, "all");
 
// Дополнительные параметры для надёжности
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
 
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processors");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
 
// Exactly-once на стороне консьюмера
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
 
// Управление offset вручную
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
consumerProps.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
 
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    if (records.isEmpty()) {
        continue;
    }
    
    producer.beginTransaction();
    
    try {
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        
        for (ConsumerRecord<String, String> record : records) {
            // Обработка
            String processed = processRecord(record);
            
            // Отправка результата в выходной топик
            producer.send(new ProducerRecord<>("output-topic", 
                record.key(), processed));
            
            // Собираем offset для коммита
            offsets.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            );
        }
        
        // Коммит offset в транзакции
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
        producer.commitTransaction();
        
    } catch (Exception e) {
        log.error("Transaction failed, aborting", e);
        producer.abortTransaction();
        // Следующий poll вернет те же записи
    }
}

public boolean wasProcessed(String eventId) {
  return redisTemplate.hasKey("processed:" + eventId);
}
 
public void markProcessed(String eventId) {
  redisTemplate.opsForValue().set(
    "processed:" + eventId, 
    "1", 
    Duration.ofDays(1)
  );
}

public void markProcessed(String eventId, ProcessingMetadata meta) {
    String value = objectMapper.writeValueAsString(meta);
    redisTemplate.opsForValue().set(
        "evt:" + eventId,
        value,
        Duration.ofHours(24)
    );
}

DB db = factory.open(new File("/data/processed"), options);
 
public boolean wasProcessed(String eventId) {
    return db.get(bytes(eventId)) != null;
}
 
public void markProcessed(String eventId) {
    db.put(bytes(eventId), bytes(System.currentTimeMillis()));
}

@Transactional
public void processInventoryUpdate(InventoryUpdateEvent event) {
    int updated = jdbcTemplate.update(
        "UPDATE inventory SET quantity = ?, version = version + 1 " +
        "WHERE item_id = ? AND version = ?",
        event.getNewQuantity(),
        event.getItemId(),
        event.getExpectedVersion()
    );
    
    if (updated == 0) {
        // Либо запись не найдена, либо версия не та - в любом случае skip
        log.debug("Skipped duplicate or outdated event for item {}", 
            event.getItemId());
    }
}

public void sendPaymentNotification(PaymentEvent event) {
    // Используем event ID как idempotency token
    HttpHeaders headers = new HttpHeaders();
    headers.set("Idempotency-Key", event.getEventId());
    
    HttpEntity<PaymentNotification> request = new HttpEntity<>(
        new PaymentNotification(event),
        headers
    );
    
    restTemplate.postForEntity(
        "https://notification-service/api/send",
        request,
        Void.class
    );
}

@Transactional  
public void processOrderEvent(OrderStatusEvent event) {
    Order order = orderRepository.findById(event.getOrderId());
    
    if (!order.canTransitionTo(event.getNewStatus())) {
        log.warn("Invalid or duplicate transition from {} to {} for order {}",
            order.getStatus(), event.getNewStatus(), order.getId());
        return;
    }
    
    order.transitionTo(event.getNewStatus());
    orderRepository.save(order);
}

public void processRefund(RefundEvent event) {
    String token = "refund:" + event.getTransactionId();
    
    if (compensationRepo.exists(token)) {
        // Уже обрабатывали, но неясно успешно ли
        // Проверяем итоговое состояние и корректируем если нужно
        reconcileRefund(event);
        return;
    }
    
    compensationRepo.save(new CompensationToken(token));
    
    try {
        walletService.addMoney(event.getUserId(), event.getAmount());
        refundRepo.recordRefund(event);
    } catch (Exception e) {
        // Rollback через compensation если можем
        compensationRepo.delete(token);
        throw e;
    }
}

while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  
  long startTime = System.currentTimeMillis();
  int processed = processRecords(records);
  long elapsed = System.currentTimeMillis() - startTime;
  
  // Если обработка заняла больше 80% от poll interval - притормози
  if (elapsed > MAX_POLL_INTERVAL * 0.8) {
      Set<TopicPartition> partitions = records.partitions();
      consumer.pause(partitions);
      log.warn("Pausing {} partitions due to slow processing", partitions.size());
      
      // Ждём чуть-чуть перед следующей попыткой
      Thread.sleep(5000);
      consumer.resume(partitions);
  }
}

private final RateLimiter rateLimiter = RateLimiter.create(1000.0); // 1000 ops/sec
 
public void processRecords(ConsumerRecords<String, String> records) {
  for (ConsumerRecord<String, String> record : records) {
      rateLimiter.acquire(); // блокируется если превышен лимит
      
      try {
          handleEvent(record);
      } catch (Exception e) {
          log.error("Failed to process event", e);
      }
  }
}

java -javaagent:opentelemetry-javaagent.jar \
     -Dotel.service.name=order-service \
     -Dotel.traces.exporter=otlp \
     -Dotel.exporter.otlp.endpoint=http://otel-collector:4317 \
     -jar application.jar

traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01

@WithSpan(value = "process.order.created")
public void processOrderCreated(OrderCreatedEvent event) {
    Span span = Span.current();
    span.setAttribute("order.id", event.getOrderId());
    span.setAttribute("order.amount", event.getTotalAmount());
    span.setAttribute("user.id", event.getUserId());
    
    try {
        // Бизнес-логика
        inventoryService.reserve(event.getItems());
        paymentService.charge(event);
        
        span.setStatus(StatusCode.OK);
    } catch (Exception e) {
        span.setStatus(StatusCode.ERROR, e.getMessage());
        span.recordException(e);
        throw e;
    }
}

private final LongCounter ordersProcessed = meter
    .counterBuilder("orders.processed")
    .setDescription("Total orders processed")
    .build();
 
private final DoubleHistogram processingDuration = meter
    .histogramBuilder("orders.processing.duration")
    .setUnit("ms")
    .build();
 
public void process(Order order) {
    long start = System.currentTimeMillis();
    try {
        // processing
        ordersProcessed.add(1, 
            Attributes.of(stringKey("status"), order.getStatus()));
    } finally {
        double duration = System.currentTimeMillis() - start;
        processingDuration.record(duration);
    }
}

@Entity
@Table(name = "orders")
public class Order {
    @Id
    @GeneratedValue
    private UUID id;
    
    @Column(name = "user_id", nullable = false)
    private String userId;
    
    @Column(name = "total_amount", nullable = false)
    private BigDecimal totalAmount;
    
    @Enumerated(EnumType.STRING)
    @Column(nullable = false)
    private OrderStatus status;
    
    @OneToMany(mappedBy = "order", cascade = CascadeType.ALL)
    private List<OrderItem> items = new ArrayList<>();
    
    @Version
    private Integer version;
    
    @Column(name = "created_at")
    private Instant createdAt = Instant.now();
    
    // конструкторы, геттеры, сеттеры
}
 
@Entity
@Table(name = "order_items")
public class OrderItem {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @ManyToOne
    @JoinColumn(name = "order_id")
    private Order order;
    
    @Column(name = "product_id", nullable = false)
    private String productId;
    
    @Column(nullable = false)
    private Integer quantity;
    
    @Column(nullable = false)
    private BigDecimal price;
}
 
public enum OrderStatus {
    CREATED, INVENTORY_RESERVED, PAYMENT_PROCESSED, 
    COMPLETED, CANCELLED
}

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderService {
    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;
    
    @Transactional
    public Order createOrder(CreateOrderRequest request) {
        // Валидация запроса
        validateRequest(request);
        
        // Создаем заказ
        Order order = new Order();
        order.setUserId(request.getUserId());
        order.setStatus(OrderStatus.CREATED);
        
        BigDecimal total = BigDecimal.ZERO;
        for (OrderItemRequest itemReq : request.getItems()) {
            OrderItem item = new OrderItem();
            item.setProductId(itemReq.getProductId());
            item.setQuantity(itemReq.getQuantity());
            item.setPrice(itemReq.getPrice());
            item.setOrder(order);
            
            order.getItems().add(item);
            total = total.add(itemReq.getPrice()
                .multiply(BigDecimal.valueOf(itemReq.getQuantity())));
        }
        order.setTotalAmount(total);
        
        // Сохраняем в базу
        Order savedOrder = orderRepository.save(order);
        log.info("Created order {} for user {}", savedOrder.getId(), order.getUserId());
        
        // Создаём событие для outbox
        OrderCreatedEvent event = new OrderCreatedEvent(
            savedOrder.getId().toString(),
            savedOrder.getUserId(),
            savedOrder.getTotalAmount(),
            savedOrder.getItems().stream()
                .map(item -> new OrderItemDto(
                    item.getProductId(),
                    item.getQuantity(),
                    item.getPrice()
                ))
                .collect(Collectors.toList())
        );
        
        OutboxEvent outboxEvent = OutboxEvent.builder()
            .aggregateId(savedOrder.getId().toString())
            .aggregateType("Order")
            .eventType("OrderCreated")
            .payload(objectMapper.writeValueAsString(event))
            .build();
        
        outboxRepository.save(outboxEvent);
        log.debug("Saved outbox event for order {}", savedOrder.getId());
        
        return savedOrder;
    }
    
    private void validateRequest(CreateOrderRequest request) {
        if (request.getItems() == null || request.getItems().isEmpty()) {
            throw new IllegalArgumentException("Order must contain at least one item");
        }
        // дополнительные проверки
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class OutboxPublisher {
    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishEvents() {
        // Берём batch с блокировкой
        List<OutboxEvent> events = outboxRepository
            .lockAndFetchUnpublished(100);
        
        if (events.isEmpty()) {
            return;
        }
        
        log.info("Publishing {} events", events.size());
        
        for (OutboxEvent event : events) {
            try {
                String topic = resolveTopicName(event.getEventType());
                
                // Синхронная отправка с ожиданием подтверждения
                SendResult<String, String> result = kafkaTemplate.send(
                    topic,
                    event.getAggregateId(),
                    event.getPayload()
                ).get(5, TimeUnit.SECONDS);
                
                // Помечаем как опубликованное только после успеха
                event.setPublished(true);
                event.setProcessedAt(Instant.now());
                outboxRepository.save(event);
                
                log.debug("Published event {} to {} at offset {}", 
                    event.getId(), 
                    topic,
                    result.getRecordMetadata().offset());
                    
            } catch (Exception e) {
                log.error("Failed to publish event {}, will retry", 
                    event.getId(), e);
                // Оставляем unpublished для следующей попытки
            }
        }
    }
    
    private String resolveTopicName(String eventType) {
        return switch (eventType) {
            case "OrderCreated" -> "order-events";
            case "OrderCancelled" -> "order-events";
            case "InventoryReserved" -> "inventory-events";
            default -> "default-events";
        };
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class InventoryConsumer {
    private final InventoryService inventoryService;
    private final ProcessedEventRepository processedEventRepo;
    private final ObjectMapper objectMapper;
    
    @KafkaListener(topics = "order-events", groupId = "inventory-service")
    @Transactional
    public void handleOrderEvent(ConsumerRecord<String, String> record) {
        String eventId = generateEventId(record);
        
        // Проверка дедупликации
        if (processedEventRepo.existsById(eventId)) {
            log.debug("Event {} already processed, skipping", eventId);
            return;
        }
        
        try {
            OrderCreatedEvent event = objectMapper.readValue(
                record.value(), 
                OrderCreatedEvent.class
            );
            
            // Резервируем товары
            boolean reserved = inventoryService.reserveItems(
                event.getOrderId(),
                event.getItems()
            );
            
            if (!reserved) {
                log.warn("Failed to reserve inventory for order {}", 
                    event.getOrderId());
                // Можно отправить компенсирующее событие
                return;
            }
            
            // Записываем ID обработанного события
            ProcessedEvent processed = new ProcessedEvent();
            processed.setEventId(eventId);
            processed.setEventType("OrderCreated");
            processed.setAggregateId(event.getOrderId());
            processedEventRepo.save(processed);
            
            log.info("Reserved inventory for order {}", event.getOrderId());
            
        } catch (Exception e) {
            log.error("Error processing event from offset {}", 
                record.offset(), e);
            throw new RuntimeException("Processing failed", e);
        }
    }
    
    private String generateEventId(ConsumerRecord<String, String> record) {
        // Комбинация топика, партиции и offset - уникальный идентификатор
        return String.format("%s-%d-%d", 
            record.topic(),
            record.partition(), 
            record.offset()
        );
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class InventoryService {
    private final InventoryRepository inventoryRepository;
    
    @Transactional
    public boolean reserveItems(String orderId, List<OrderItemDto> items) {
        for (OrderItemDto item : items) {
            Inventory inventory = inventoryRepository
                .findByProductId(item.getProductId())
                .orElseThrow(() -> new ProductNotFoundException(
                    item.getProductId()));
            
            if (inventory.getAvailableQuantity() < item.getQuantity()) {
                log.warn("Insufficient inventory for product {}", 
                    item.getProductId());
                return false;
            }
            
            // Уменьшаем доступное количество
            inventory.setAvailableQuantity(
                inventory.getAvailableQuantity() - item.getQuantity());
            inventory.setReservedQuantity(
                inventory.getReservedQuantity() + item.getQuantity());
            
            inventoryRepository.save(inventory);
            
            log.debug("Reserved {} units of product {} for order {}", 
                item.getQuantity(), 
                item.getProductId(),
                orderId);
        }
        
        return true;
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class NotificationConsumer {
    private final RedisTemplate<String, String> redisTemplate;
    private final NotificationService notificationService;
    private final ObjectMapper objectMapper;
    
    @KafkaListener(topics = "order-events", groupId = "notification-service")
    public void handleOrderEvent(ConsumerRecord<String, String> record) {
        String eventKey = "notif:" + record.topic() + ":" + 
            record.partition() + ":" + record.offset();
        
        // Проверяем Redis на дубликат
        Boolean wasProcessed = redisTemplate.opsForValue()
            .setIfAbsent(eventKey, "1", Duration.ofHours(24));
        
        if (Boolean.FALSE.equals(wasProcessed)) {
            log.debug("Notification already sent for event at offset {}", 
                record.offset());
            return;
        }
        
        try {
            OrderCreatedEvent event = objectMapper.readValue(
                record.value(),
                OrderCreatedEvent.class
            );
            
            // Отправка уведомления с idempotency token
            notificationService.sendOrderConfirmation(
                event.getUserId(),
                event.getOrderId(),
                event.getTotalAmount(),
                eventKey  // используем как idempotency key
            );
            
            log.info("Sent notification for order {} to user {}", 
                event.getOrderId(), 
                event.getUserId());
                
        } catch (Exception e) {
            // Удаляем ключ чтобы retry сработал
            redisTemplate.delete(eventKey);
            log.error("Failed to send notification, will retry", e);
            throw new RuntimeException("Notification failed", e);
        }
    }
}

@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
@Slf4j
public class OrderController {
    private final OrderService orderService;
    private final OrderRepository orderRepository;
    
    @PostMapping
    public ResponseEntity<OrderResponse> createOrder(
            @RequestBody @Valid CreateOrderRequest request) {
        try {
            Order order = orderService.createOrder(request);
            
            OrderResponse response = OrderResponse.builder()
                .orderId(order.getId().toString())
                .status(order.getStatus().name())
                .totalAmount(order.getTotalAmount())
                .createdAt(order.getCreatedAt())
                .build();
            
            return ResponseEntity.status(HttpStatus.CREATED).body(response);
            
        } catch (IllegalArgumentException e) {
            log.warn("Invalid order request: {}", e.getMessage());
            return ResponseEntity.badRequest().build();
        } catch (Exception e) {
            log.error("Failed to create order", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    @GetMapping("/{orderId}")
    public ResponseEntity<OrderResponse> getOrder(@PathVariable String orderId) {
        return orderRepository.findById(UUID.fromString(orderId))
            .map(order -> {
                OrderResponse response = OrderResponse.builder()
                    .orderId(order.getId().toString())
                    .userId(order.getUserId())
                    .status(order.getStatus().name())
                    .totalAmount(order.getTotalAmount())
                    .items(order.getItems().stream()
                        .map(item -> new OrderItemDto(
                            item.getProductId(),
                            item.getQuantity(),
                            item.getPrice()
                        ))
                        .collect(Collectors.toList()))
                    .createdAt(order.getCreatedAt())
                    .build();
                
                return ResponseEntity.ok(response);
            })
            .orElse(ResponseEntity.notFound().build());
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class ErrorHandlingConsumer {
    private final KafkaTemplate<String, String> dlqTemplate;
    private final RedisTemplate<String, String> retryTracker;
    
    private static final int MAX_RETRIES = 3;
    private static final String DLQ_TOPIC = "order-events-dlq";
    
    public void handleWithRetry(ConsumerRecord<String, String> record,
                                 Consumer<ConsumerRecord<String, String>> processor) {
        String retryKey = "retry:" + record.topic() + ":" + 
            record.partition() + ":" + record.offset();
        
        // Считаем количество попыток
        String retriesStr = retryTracker.opsForValue().get(retryKey);
        int retries = retriesStr != null ? Integer.parseInt(retriesStr) : 0;
        
        try {
            processor.accept(record);
            
            // Успех - удаляем счетчик
            retryTracker.delete(retryKey);
            
        } catch (Exception e) {
            retries++;
            
            if (retries >= MAX_RETRIES) {
                log.error("Max retries exceeded for offset {}, sending to DLQ",
                    record.offset(), e);
                
                // Отправляем в Dead Letter Queue
                sendToDLQ(record, e);
                retryTracker.delete(retryKey);
                
            } else {
                log.warn("Retry {}/{} for offset {}", 
                    retries, MAX_RETRIES, record.offset());
                
                // Сохраняем счетчик с TTL
                retryTracker.opsForValue().set(
                    retryKey, 
                    String.valueOf(retries),
                    Duration.ofHours(1)
                );
                
                // Пробрасываем исключение для retry
                throw e;
            }
        }
    }
    
    private void sendToDLQ(ConsumerRecord<String, String> record, Exception error) {
        DLQMessage dlqMsg = new DLQMessage(
            record.topic(),
            record.partition(),
            record.offset(),
            record.key(),
            record.value(),
            error.getClass().getName(),
            error.getMessage(),
            Instant.now()
        );
        
        try {
            dlqTemplate.send(DLQ_TOPIC, 
                objectMapper.writeValueAsString(dlqMsg));
        } catch (Exception e) {
            log.error("Failed to send to DLQ", e);
        }
    }
}

@Component
@RequiredArgsConstructor
public class MetricsCollector {
    private final MeterRegistry registry;
    
    private final Counter ordersCreated;
    private final Counter eventsPublished;
    private final Timer eventPublishLatency;
    private final Gauge outboxQueueSize;
    
    @PostConstruct
    public void init() {
        ordersCreated = Counter.builder("orders.created")
            .description("Total orders created")
            .register(registry);
        
        eventsPublished = Counter.builder("outbox.events.published")
            .description("Events successfully published to Kafka")
            .register(registry);
        
        eventPublishLatency = Timer.builder("outbox.publish.latency")
            .description("Time to publish event from outbox")
            .register(registry);
    }
    
    public void recordOrderCreated() {
        ordersCreated.increment();
    }
    
    public void recordEventPublished(long latencyMs) {
        eventsPublished.increment();
        eventPublishLatency.record(latencyMs, TimeUnit.MILLISECONDS);
    }
}

@SpringBootTest
@Testcontainers
class OrderFlowIntegrationTest {
    
    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15")
        .withDatabaseName("test_db");
    
    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.5.0"));
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private OutboxEventRepository outboxRepository;
    
    @Autowired
    private OrderRepository orderRepository;
    
    @DynamicPropertySource
    static void properties(DynamicPropertyRegistry registry) {
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }
    
    @Test
    void shouldCreateOrderAndPublishEvent() throws Exception {
        // Given
        CreateOrderRequest request = CreateOrderRequest.builder()
            .userId("user-123")
            .items(List.of(
                new OrderItemRequest("product-1", 2, new BigDecimal("10.00")),
                new OrderItemRequest("product-2", 1, new BigDecimal("25.50"))
            ))
            .build();
        
        // When - создаём заказ
        Order order = orderService.createOrder(request);
        
        // Then - заказ в базе
        assertThat(order.getId()).isNotNull();
        assertThat(order.getStatus()).isEqualTo(OrderStatus.CREATED);
        assertThat(order.getTotalAmount())
            .isEqualByComparingTo(new BigDecimal("45.50"));
        
        // And - событие в outbox
        List<OutboxEvent> events = outboxRepository
            .findByAggregateId(order.getId().toString());
        assertThat(events).hasSize(1);
        
        OutboxEvent event = events.get(0);
        assertThat(event.getEventType()).isEqualTo("OrderCreated");
        assertThat(event.isPublished()).isFalse();
        
        // When - publisher обрабатывает outbox
        outboxPublisher.publishEvents();
        
        // Then - событие опубликовано
        OutboxEvent published = outboxRepository.findById(event.getId()).get();
        assertThat(published.isPublished()).isTrue();
        assertThat(published.getProcessedAt()).isNotNull();
    }
    
    @Test
    void shouldHandleDuplicateOrderCreation() {
        CreateOrderRequest request = createTestRequest();
        
        // Создаём заказ дважды
        Order order1 = orderService.createOrder(request);
        Order order2 = orderService.createOrder(request);
        
        // Оба должны создаться (разные заказы)
        assertThat(order1.getId()).isNotEqualTo(order2.getId());
        
        // Но если консьюмер обработает одно событие дважды
        // дедупликация должна сработать
    }
}

@Component
public class KafkaHealthIndicator implements HealthIndicator {
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    @Override
    public Health health() {
        try {
            // Пробуем получить metadata от Kafka
            kafkaTemplate.partitionsFor("order-events");
            return Health.up()
                .withDetail("kafka", "Connected")
                .build();
        } catch (Exception e) {
            return Health.down()
                .withDetail("kafka", "Unavailable")
                .withException(e)
                .build();
        }
    }
}