Spring Kafka - синхронные и асинхронные запросы

// Псевдокод операции, требующей синхронного выполнения
Результат result = serviceA.выполнитьОперацию();
if (result.успешно) {
    serviceB.выполнитьЗависимуюОперацию(result.данные);
} else {
    обработатьОшибку(result.ошибка);
}

// Синхронный подход с сильной согласованностью
String response = replyingKafkaTemplate.sendAndReceive(record).get().value();
// Поток блокируется до получения ответа
 
// Асинхронный подход с высокой производительностью
kafkaTemplate.send(record);
// Поток сразу продолжает выполнение

// Привязка сообщений к конкретной партиции через ключ
ProducerRecord<String, String> record = new ProducerRecord<>(
    topicName,
    userId, // Ключ для партиционирования
    messagePayload
);

// Отправка синхронного запроса
ProducerRecord<String, String> record = new ProducerRecord<>(requestTopic, "Запрос данных");
record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, replyTopic.getBytes()));
 
RequestReplyFuture<String, String, String> future = replyingTemplate.sendAndReceive(record);
ConsumerRecord<String, String> response = future.get(30, TimeUnit.SECONDS);
String result = response.value();

@Configuration
public class KafkaConfig {
 
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    public static final String REQUEST_TOPIC = "request-topic";
    public static final String REPLY_TOPIC = "reply-topic";
    
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(config);
    }
    
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "replying-group");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // Важно для запрос-ответ паттерна, чтобы быстро получать ответы
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        return new DefaultKafkaConsumerFactory<>(config);
    }
    
    @Bean
    public ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate(
            ProducerFactory<String, String> pf,
            KafkaMessageListenerContainer<String, String> container) {
        ReplyingKafkaTemplate<String, String, String> template = 
            new ReplyingKafkaTemplate<>(pf, container);
        // Установка таймаута ожидания ответа
        template.setDefaultReplyTimeout(Duration.ofSeconds(30));
        return template;
    }
    
    @Bean
    public KafkaMessageListenerContainer<String, String> replyContainer(
            ConsumerFactory<String, String> cf) {
        ContainerProperties props = new ContainerProperties(REPLY_TOPIC);
        return new KafkaMessageListenerContainer<>(cf, props);
    }
}

@Bean
public ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate(
        ProducerFactory<String, String> pf,
        KafkaMessageListenerContainer<String, String> container) {
    ReplyingKafkaTemplate<String, String, String> template = 
        new ReplyingKafkaTemplate<>(pf, container);
    template.setDefaultReplyTimeout(Duration.ofSeconds(30));
    
    // Настройка очистки просроченных запросов
    template.setSharedReplyTopic(true);
    template.setReplyTopicCheckInterval(1000);
    
    return template;
}

template.setReplyTimeout(Duration.ofSeconds(30));
// Максимум 100 одновременных запросов
template.setMaxInFlight(100);

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    // Базовые настройки...
    
    // Включение транзакций
    config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-");
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    
    return new DefaultKafkaProducerFactory<>(config);
}

@Value("${kafka.bootstrap-servers}")
private String bootstrapServers;
 
private Map<String, Object> baseProducerConfig() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    // Другие параметры...
    return props;
}

// Настройки повторных попыток
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000);
 
// Настройки таймаутов
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 15000);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 30000);

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

@Bean
public ProducerFactory<String, OrderRequest> orderRequestProducerFactory() {
    Map<String, Object> props = baseProducerConfig();
    
    // Для типизированной фабрики
    DefaultKafkaProducerFactory<String, OrderRequest> factory = 
        new DefaultKafkaProducerFactory<>(props);
    
    factory.setValueSerializer(new JsonSerializer<>(new TypeReference<OrderRequest>() {}));
    return factory;
}

@Bean
public ReplyingKafkaTemplate<String, OrderRequest, ValidationResult> replyingTemplate(
        ProducerFactory<String, OrderRequest> pf,
        KafkaMessageListenerContainer<String, ValidationResult> container) {
    return new ReplyingKafkaTemplate<>(pf, container);
}
 
@Bean
public KafkaMessageListenerContainer<String, ValidationResult> replyContainer(
        ConsumerFactory<String, ValidationResult> cf) {
    ContainerProperties props = new ContainerProperties(REPLY_TOPIC);
    return new KafkaMessageListenerContainer<>(cf, props);
}
 
@Bean
public ConsumerFactory<String, ValidationResult> validationResultConsumerFactory() {
    Map<String, Object> props = baseConsumerConfig();
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.mycompany.model");
    return new DefaultKafkaConsumerFactory<>(props, 
        new StringDeserializer(), 
        new JsonDeserializer<>(ValidationResult.class));
}

@Service
public class KafkaRequestClient {
 
    @Autowired
    private ReplyingKafkaTemplate<String, RequestDto, ResponseDto> replyingKafkaTemplate;
 
    public ResponseDto sendRequest(RequestDto request) throws Exception {
        // Создаем запись с топиком запроса
        ProducerRecord<String, RequestDto> record = new ProducerRecord<>(
                "request-topic", request);
        
        // Добавляем заголовок с именем топика для ответа
        record.headers().add(
            new RecordHeader(KafkaHeaders.REPLY_TOPIC, "reply-topic".getBytes())
        );
        
        // Отправляем запрос и ждем ответа
        RequestReplyFuture<String, RequestDto, ResponseDto> future = 
            replyingKafkaTemplate.sendAndReceive(record);
        
        // Блокируем поток до получения ответа или таймаута
        ConsumerRecord<String, ResponseDto> response = future.get(30, TimeUnit.SECONDS);
        
        return response.value();
    }
}

@Service
public class RequestProcessor {
 
    @KafkaListener(topics = "request-topic", groupId = "request-processors")
    @SendTo // Магия Spring - возвращаемое значение отправится в reply-topic
    public ResponseDto processRequest(RequestDto request) {
        // Бизнес-логика обработки запроса
        ResponseDto response = new ResponseDto();
        response.setStatus("SUCCESS");
        response.setProcessedData(doSomethingWith(request));
        
        return response; // Будет автоматически отправлено как ответ
    }
    
    private String doSomethingWith(RequestDto request) {
        // Реальная бизнес-логика
        return "Processed: " + request.getData();
    }
}

@KafkaListener(topics = "request-topic")
public ResponseDto processWithCustomReplyTopic(
        @Payload RequestDto request,
        @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTopicBytes,
        @Header(KafkaHeaders.CORRELATION_ID) byte[] correlationId,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic,
        KafkaTemplate<String, ResponseDto> kafkaTemplate) {
    
    String replyTopic = new String(replyTopicBytes, StandardCharsets.UTF_8);
    
    ResponseDto response = new ResponseDto();
    // ... бизнес-логика ...
    
    // Ручная отправка ответа
    ProducerRecord<String, ResponseDto> record = 
        new ProducerRecord<>(replyTopic, response);
    record.headers().add(new RecordHeader(KafkaHeaders.CORRELATION_ID, correlationId));
    
    kafkaTemplate.send(record);
    
    return null; // Не используем @SendTo
}

@KafkaListener(topics = "request-topic")
@SendTo
public CompletableFuture<ResponseDto> processRequestAsync(RequestDto request) {
    return CompletableFuture.supplyAsync(() -> {
        // Длительная операция в отдельном потоке
        ResponseDto response = new ResponseDto();
        // ... логика ...
        return response;
    });
}

// Опасный код - без ограничения количества запросов
@GetMapping("/data/{id}")
public ResponseEntity<DataDto> getData(@PathVariable String id) throws Exception {
    // Блокирующий вызов без контроля количества одновременных запросов
    DataDto result = replyingKafkaTemplate.sendAndReceive(
        new ProducerRecord<>("request-topic", id)).get().value();
    return ResponseEntity.ok(result);
}

private final Semaphore requestLimiter = new Semaphore(50); // Максимум 50 одновременных запросов
 
public ResponseDto sendWithLimiting(RequestDto request) throws Exception {
    boolean acquired = requestLimiter.tryAcquire(100, TimeUnit.MILLISECONDS);
    if (!acquired) {
        throw new ServiceUnavailableException("Сервис перегружен, попробуйте позже");
    }
    
    try {
        // Блокирующий вызов, но с ограничением по количеству
        return replyingKafkaTemplate.sendAndReceive(
            new ProducerRecord<>("request-topic", request)).get(5, TimeUnit.SECONDS).value();
    } finally {
        requestLimiter.release(); // Обязательно освобождаем семафор
    }
}

// Пример с разными уровнями таймаутов
ReplyingKafkaTemplate<String, String, String> template = new ReplyingKafkaTemplate<>(pf, container);
// Таймаут на уровне шаблона
template.setDefaultReplyTimeout(Duration.ofSeconds(10));
 
// ...
 
// В методе сервиса - таймаут на уровне вызова
try {
    return future.get(5, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    // Обработка таймаута
}

public ResponseDto sendWithRetry(RequestDto request) {
    int maxRetries = 3;
    int attempt = 0;
    long delay = 100; // начальная задержка в мс
    
    while (attempt < maxRetries) {
        try {
            return replyingKafkaTemplate.sendAndReceive(
                new ProducerRecord<>("request-topic", request))
                .get(5, TimeUnit.SECONDS).value();
        } catch (TimeoutException e) {
            attempt++;
            if (attempt >= maxRetries) throw new ServiceException("Максимальное число попыток исчерпано");
            try {
                Thread.sleep(delay);
                delay *= 2; // экспоненциальное увеличение задержки
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                throw new ServiceException("Прервано во время ожидания повтора");
            }
        }
    }
    throw new ServiceException("Не удалось получить ответ после повторных попыток");
}

@Bean
public MeterRegistry meterRegistry() {
    return new SimpleMeterRegistry();
}
 
@Bean
public ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate(
        ProducerFactory<String, String> pf,
        KafkaMessageListenerContainer<String, String> container,
        MeterRegistry meterRegistry) {
    ReplyingKafkaTemplate<String, String, String> template = 
        new ReplyingKafkaTemplate<>(pf, container);
    template.setMicrometerEnabled(true);
    template.setMeterRegistry(meterRegistry);
    return template;
}

private final Counter timeoutCounter;
 
public KafkaRequestService(MeterRegistry registry) {
    this.timeoutCounter = registry.counter("kafka.request.timeouts");
}
 
public ResponseDto sendRequest(RequestDto request) {
    try {
        return replyingTemplate.sendAndReceive(record).get(5, TimeUnit.SECONDS).value();
    } catch (TimeoutException e) {
        timeoutCounter.increment();
        throw new ServiceException("Таймаут при обработке запроса");
    }
}

@Scheduled(fixedRate = 10000) // каждые 10 секунд
public void checkPendingReplies() {
    int pendingReplies = replyingTemplate.getPendingReplyCount();
    pendingRepliesGauge.set(pendingReplies);
    
    if (pendingReplies > 100) {
        log.warn("Большое количество ожидающих ответов: {}", pendingReplies);
    }
}

@KafkaListener(topics = "request-topic", errorHandler = "kafkaErrorHandler")
@SendTo
public ResponseDto processRequest(RequestDto request) {
    // Обработка запроса, которая может выбросить исключение
    return processRequestLogic(request);
}
 
@Bean
public KafkaListenerErrorHandler kafkaErrorHandler(KafkaTemplate<String, Object> kafkaTemplate) {
    return (message, exception) -> {
        // Записываем проблемное сообщение в DLQ
        ProducerRecord<String, Object> deadRecord = 
            new ProducerRecord<>("dead-letter-topic", message.getPayload());
        
        // Копируем важные заголовки
        message.getHeaders().forEach((key, value) -> {
            if (key.equals(KafkaHeaders.CORRELATION_ID) || 
                key.equals(KafkaHeaders.REPLY_TOPIC)) {
                deadRecord.headers().add(new RecordHeader(key, (byte[]) value));
            }
        });
        
        // Добавляем информацию об ошибке
        deadRecord.headers().add(
            new RecordHeader("exception", exception.getMessage().getBytes())
        );
        
        kafkaTemplate.send(deadRecord);
        
        // Возвращаем ответ с ошибкой, чтобы клиент не ждал таймаута
        ErrorResponseDto errorResponse = new ErrorResponseDto();
        errorResponse.setStatus("ERROR");
        errorResponse.setMessage("Запрос не может быть обработан");
        
        return errorResponse;
    };
}

@Bean
public KafkaAdmin.NewTopics deadLetterTopics() {
    return new KafkaAdmin.NewTopics(
        TopicBuilder.name("dead-letter-topic")
            .partitions(3)
            .replicas(2)
            .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(TimeUnit.DAYS.toMillis(7)))
            .build()
    );
}

// Отправка асинхронного сообщения
kafkaTemplate.send("request-topic", requestMessage)
    .addCallback(
        result -> log.info("Сообщение успешно отправлено: {}", result),
        ex -> log.error("Ошибка отправки сообщения", ex)
    );
 
// Код продолжает выполнение, не ожидая ответа

@KafkaListener(topics = "response-topic-for-service-a")
public void handleResponse(ConsumerRecord<String, ResponseDto> record) {
    String correlationId = new String(
        record.headers().lastHeader(KafkaHeaders.CORRELATION_ID).value()
    );
    
    // Находим ожидающий запрос по correlationId и обрабатываем ответ
    pendingRequests.complete(correlationId, record.value());
}

// Хранилище ожидающих запросов
private final ConcurrentHashMap<String, CompletableFuture<ResponseDto>> pendingRequests = 
    new ConcurrentHashMap<>();
 
public CompletableFuture<ResponseDto> sendAsyncRequest(RequestDto request) {
    // Генерируем уникальный идентификатор корреляции
    String correlationId = UUID.randomUUID().toString();
    
    // Создаем CompletableFuture, который будет заполнен позже
    CompletableFuture<ResponseDto> future = new CompletableFuture<>();
    pendingRequests.put(correlationId, future);
    
    // Настраиваем автоматическое удаление по таймауту
    scheduledExecutor.schedule(() -> {
        CompletableFuture<ResponseDto> pending = pendingRequests.remove(correlationId);
        if (pending != null) {
            pending.completeExceptionally(
                new TimeoutException("Таймаут ожидания ответа")
            );
        }
    }, 30, TimeUnit.SECONDS);
    
    // Отправляем сообщение с идентификатором корреляции
    ProducerRecord<String, RequestDto> record = 
        new ProducerRecord<>("request-topic", request);
    record.headers().add(
        new RecordHeader(KafkaHeaders.CORRELATION_ID, correlationId.getBytes())
    );
    record.headers().add(
        new RecordHeader(KafkaHeaders.REPLY_TOPIC, "response-topic".getBytes())
    );
    
    // Отправляем без блокировки
    kafkaTemplate.send(record);
    
    return future; // Возвращаем сразу, не дожидаясь ответа
}

@KafkaListener(topics = "response-topic")
public void handleResponse(
    @Payload ResponseDto response,
    @Header(KafkaHeaders.CORRELATION_ID) byte[] correlationIdBytes) {
    
    String correlationId = new String(correlationIdBytes, StandardCharsets.UTF_8);
    CompletableFuture<ResponseDto> future = pendingRequests.remove(correlationId);
    
    if (future != null) {
        future.complete(response); // Заполняем future результатом
    } else {
        log.warn("Получен ответ для неизвестного запроса: {}", correlationId);
    }
}

// Вариант 1: блокирующее ожидание, когда это действительно нужно
ResponseDto response = kafkaService.sendAsyncRequest(request).get(10, TimeUnit.SECONDS);
 
// Вариант 2: обработка результата, когда он будет доступен
kafkaService.sendAsyncRequest(request)
    .thenAccept(result -> processResult(result))
    .exceptionally(ex -> {
        handleError(ex);
        return null;
    });
 
// Вариант 3: композиция нескольких запросов
CompletableFuture<UserDto> userFuture = kafkaService.sendAsyncRequest(userRequest);
CompletableFuture<OrderDto> orderFuture = kafkaService.sendAsyncRequest(orderRequest);
 
CompletableFuture.allOf(userFuture, orderFuture)
    .thenRun(() -> {
        UserDto user = userFuture.join();
        OrderDto order = orderFuture.join();
        combineResults(user, order);
    });

@KafkaListener(topics = "my-topic", groupId = "my-group")
public void processMessage(String message) {
    log.info("Получено сообщение: {}", message);
    // Обработка сообщения
}

@KafkaListener(topics = "my-topic")
public void processWithMetadata(
    @Payload String content,
    @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
    @Header(KafkaHeaders.OFFSET) long offset,
    @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp,
    ConsumerRecord<String, String> record) {
    
    log.info("Получено сообщение из партиции {} с offset {}", partition, offset);
    // Доступ к заголовкам через record.headers()
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> customContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    
    // Настройка параллельной обработки
    factory.setConcurrency(10); // 10 потоков обработки
    
    // Размер пакета обрабатываемых сообщений
    factory.setBatchListener(true);
    
    // Настройка политики перезапуска в случае ошибок
    factory.setErrorHandler((exception, data) -> {
        log.error("Ошибка обработки: {}", exception.getMessage());
    });
    
    return factory;
}
 
@KafkaListener(
    topics = "high-volume-topic", 
    containerFactory = "customContainerFactory"
)
public void processBatch(List<String> messages) {
    log.info("Получен пакет из {} сообщений", messages.size());
    // Обработка пакета сообщений
}

factory.setConcurrency(10); // 10 параллельных потребителей

factory.getContainerProperties().setPollTimeout(5000);
factory.setBatchErrorHandler(new BatchLoggingErrorHandler());

@KafkaListener(topics = "critical-topic")
public void processWithManualCommit(
    @Payload String data,
    Acknowledgment ack) {
    
    try {
        // Обработка данных
        processData(data);
        
        // Подтверждение успешной обработки
        ack.acknowledge();
    } catch (Exception e) {
        log.error("Ошибка обработки, смещение не будет зафиксировано", e);
        // Здесь можно решить - пробросить исключение для повторной обработки
        // или проглотить и подтвердить
    }
}

kafkaTemplate.send(topicName, message)
    .addCallback(
        result -> {
            // Код, выполняемый при успешной отправке
            RecordMetadata metadata = result.getRecordMetadata();
            log.info("Сообщение отправлено в топик {}, партиция {}, offset {}",
                    metadata.topic(), metadata.partition(), metadata.offset());
        },
        ex -> {
            // Код, выполняемый при ошибке
            log.error("Ошибка отправки сообщения", ex);
            // Здесь можно реализовать логику повторной отправки или уведомления
        }
    );

public void logUserAction(UserAction action) {
    kafkaTemplate.send("audit-log", action)
        .addCallback(
            success -> auditSuccessCounter.increment(),
            failure -> {
                auditFailureCounter.increment();
                // Сохраняем в локальный буфер для последующей повторной отправки
                retryBuffer.add(action);
            }
        );
}

class AuditCallback implements ListenableFutureCallback<SendResult<String, UserAction>> {
    @Override
    public void onSuccess(SendResult<String, UserAction> result) {
        // Успешная отправка
    }
    
    @Override
    public void onFailure(Throwable ex) {
        // Обработка ошибки
    }
}
 
// Использование
kafkaTemplate.send("topic", message).addCallback(new AuditCallback());

public void processOrderAsync(Order order) {
    String correlationId = UUID.randomUUID().toString();
    
    // Регистрируем обработчик для конкретного запроса
    responseHandlers.put(correlationId, response -> {
        // Этот код выполнится, когда придет ответ с нужным correlationId
        notifyUserAboutOrderStatus(order.getUserId(), response.getStatus());
    });
    
    // Отправляем запрос с идентификатором корреляции
    ProducerRecord<String, Order> record = new ProducerRecord<>("orders", order);
    record.headers().add(new RecordHeader("correlation-id", correlationId.getBytes()));
    kafkaTemplate.send(record);
}

scheduledExecutor.schedule(() -> {
    responseHandlers.remove(correlationId);
}, 5, TimeUnit.MINUTES);

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    // Базовые настройки
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    
    // Настройки пула соединений
    props.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, 180000); // 3 минуты
    props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
    props.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, 30000); // 30 секунд
    return props;
}

props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000); // 3 секунды
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000); // 15 секунд

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(10); // 10 потоков обработки
    return factory;
}

// Классический пример Fire and Forget
kafkaTemplate.send("notification-topic", new UserNotification(userId, "Ваш заказ отправлен"));
// Код продолжает выполняться сразу после отправки

// Асинхронный Request-Reply с CompletableFuture
CompletableFuture<ValidationResult> future = validationService.validateAsync(order);
future.thenAccept(result -> {
    if (result.isValid()) {
        proceedWithOrder(order);
    } else {
        notifyValidationFailed(order, result.getErrors());
    }
});
// Код продолжает выполняться, не дожидаясь результата валидации

@Bean
public ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate(
        ReactiveKafkaConsumerFactory<String, String> factory) {
    return new ReactiveKafkaConsumerTemplate<>(factory);
}
 
@Bean
public ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate(
        ReactiveKafkaProducerFactory<String, String> factory) {
    return new ReactiveKafkaProducerTemplate<>(factory);
}

@Service
public class ReactiveKafkaConsumer {
 
    private final ReactiveKafkaConsumerTemplate<String, String> template;
    
    // ... конструктор ...
    
    public Flux<String> consumeMessages() {
        return template.receiveAutoAck()
            .limitRate(100) // Берем не больше 100 сообщений за раз
            .map(ConsumerRecord::value)
            .doOnNext(value -> log.info("Обработка: {}", value))
            .onErrorContinue((err, obj) -> log.error("Ошибка: {}", err.getMessage()));
    }
}

template.receiveAutoAck()
    .concatMap(record -> processRecordReactively(record)
                .doOnError(e -> log.error("Ошибка обработки", e))
                .onErrorResume(e -> Mono.empty())
    )
    .subscribe();

@Bean
public NewTopic highThroughputTopic() {
    return TopicBuilder.name("high-load-topic")
        .partitions(30)  // Оптимизировано под кластер из 10 нод
        .replicas(3)
        .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(TimeUnit.DAYS.toMillis(1)))
        .build();
}

try {
    ResponseDto response = replyingKafkaTemplate.sendAndReceive(record).get(10, TimeUnit.SECONDS);
    // Обработка успешного ответа
} catch (ExecutionException ex) {
    // Ошибка на стороне отправки или обработки
    log.error("Ошибка обработки запроса", ex.getCause());
} catch (TimeoutException ex) {
    // Таймаут ожидания ответа
    log.error("Превышено время ожидания ответа");
}

kafkaService.sendAsyncRequest(request)
    .thenApply(this::processResponse)
    .exceptionally(ex -> {
        if (ex instanceof CompletionException && ex.getCause() != null) {
            handleSpecificError(ex.getCause());
        }
        return fallbackResponse(); // Возвращаем запасной ответ вместо ошибки
    })
    .thenAccept(this::sendToClient);

private final CircuitBreaker circuitBreaker = CircuitBreaker.builder()
    .failureRateThreshold(50) // 50% ошибок
    .waitDurationInOpenState(Duration.ofMinutes(1))
    .build();
 
public ResponseDto sendWithCircuitBreaker(RequestDto request) {
    return circuitBreaker.executeSupplier(() -> 
        replyingTemplate.sendAndReceive(record).get().value()
    );
}

reactiveTemplate.receiveAutoAck()
    .flatMap(record -> processRecord(record)
        .onErrorResume(ex -> {
            log.error("Ошибка обработки: {}", ex.getMessage());
            return Mono.empty(); // Пропускаем проблемное сообщение
        })
    )
    .subscribe();

executor.setUncaughtExceptionHandler((thread, throwable) -> 
    log.error("Неперехваченное исключение в потоке {}: {}", 
             thread.getName(), throwable.getMessage())
);

// Этот код может вызвать проблемы при высокой нагрузке
for (Request request : requests) {
    Response response = kafkaTemplate.sendAndReceive(request).get(); // Блокирует поток
    process(response);
}

@Retryable(value = KafkaException.class, maxAttempts = 3, 
    backoff = @Backoff(delay = 1000, multiplier = 2))
public ResponseDto sendWithRetry(RequestDto request) {
    return replyingTemplate.sendAndReceive(record).get(5, TimeUnit.SECONDS).value();
}

private CompletableFuture<ResponseDto> sendWithAsyncRetry(RequestDto request, int attempts) {
    return kafkaTemplate.sendDefault(request)
        .thenCompose(result -> listenForResponse(result.getRecordMetadata().offset()))
        .exceptionally(ex -> {
            if (attempts > 1) {
                return CompletableFuture.delayedExecutor(1000, TimeUnit.MILLISECONDS)
                    .supplyAsync(() -> sendWithAsyncRetry(request, attempts - 1).join());
            }
            throw new CompletionException(ex);
        });
}

@Bean
public DeadLetterPublishingRecoverer recoverer(KafkaTemplate<?, ?> template) {
    return new DeadLetterPublishingRecoverer(template,
        (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));
}

reactiveKafkaTemplate.receiveAutoAck()
    .doOnNext(this::processRecord)
    .doOnError(e -> {
        // Отправка в DLT с сохранением контекста ошибки
        sendToDlt(record, e);
    });

@KafkaListener(topics = "payment-topic")
public void processPayment(PaymentEvent event) {
    // Идемпотентная операция - повторное применение безопасно
    paymentRepository.save(new Payment(
        event.getPaymentId(), // Уникальный идентификатор платежа
        event.getAmount(),
        event.getStatus()
    ));
}

public boolean isDuplicate(String messageId) {
    // NX - установить, только если не существует
    // EX - с истечением срока через 24 часа
    Boolean isSet = redisTemplate.opsForValue()
        .setIfAbsent("msg:" + messageId, "1", Duration.ofHours(24));
    
    return isSet == null || !isSet; // true если дубликат
}

try {
    // Уникальный ключ в БД предотвратит дублирование
    orderRepository.saveWithUniqueConstraint(order);
} catch (DuplicateKeyException ex) {
    log.info("Дубликат заказа проигнорирован: {}", order.getOrderId());
}

@Transactional
public void updateMetric(MetricEvent event) {
    Metric current = repository.findById(event.getMetricId()).orElse(null);
    if (current == null || event.getVersion() > current.getVersion()) {
        repository.save(new Metric(event));
    }
}

// Публикация события об изменении статуса заказа
OrderStatusChangedEvent event = new OrderStatusChangedEvent(
    orderId, 
    OrderStatus.SHIPPED,
    LocalDateTime.now()
);
kafkaTemplate.send("order-events", event);

// Обработка заказов с агрегацией данных от разных сервисов
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");
 
orders.join(customers, 
    (order, customer) -> enrichOrderWithCustomerData(order, customer))
    .to("enriched-orders");

@KafkaListener(topics = "order-events")
public void processOrderEvent(OrderEvent event) {
    // Преобразуем событие Kafka в HTTP-запрос
    ExternalApiRequest request = mapToExternalRequest(event);
    
    // Вызываем внешний API
    try {
        ExternalApiResponse response = restTemplate.postForObject(
            apiUrl, request, ExternalApiResponse.class);
        processResponse(response, event);
    } catch (RestClientException ex) {
        handleApiError(ex, event);
    }
}

@Retryable(value = RestClientException.class, 
           maxAttempts = 3, 
           backoff = @Backoff(delay = 1000, multiplier = 2))
public ExternalApiResponse callExternalApi(ExternalApiRequest request) {
    return circuitBreaker.run(
        () -> restTemplate.postForObject(apiUrl, request, ExternalApiResponse.class),
        throwable -> fallbackResponse(request, throwable)
    );
}

@KafkaListener(topics = "high-volume-events")
public void processEvent(KafkaEvent event) {
    webClient.post()
        .uri(apiUrl)
        .bodyValue(mapToRequest(event))
        .retrieve()
        .bodyToMono(ApiResponse.class)
        .doOnSuccess(this::handleSuccess)
        .doOnError(error -> handleError(error, event))
        .subscribe();
}

private final RateLimiter rateLimiter = RateLimiter.create(50.0); // 50 запросов в секунду
 
public ExternalApiResponse callWithRateLimit(ExternalApiRequest request) {
    rateLimiter.acquire(); // Блокирует, если достигнут лимит
    return restTemplate.postForObject(apiUrl, request, ExternalApiResponse.class);
}

@Bean
public KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry(
        KafkaProperties properties, MeterRegistry registry) {
    // Регистрируем метрики для всех слушателей Kafka
    properties.buildConsumerProperties()
            .forEach((k, v) -> Metrics.gauge("kafka.consumer." + k, 
                     Collections.emptyList(), v, Object::toString));
    return new KafkaListenerEndpointRegistry();
}

@KafkaListener(topics = "order-events")
public void processOrder(OrderEvent event, 
                        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key) {
    MDC.put("messageId", event.getEventId());
    MDC.put("topic", "order-events");
    MDC.put("key", key);
    
    try {
        log.info("Получено сообщение: {}", event);
        // Обработка сообщения
    } finally {
        MDC.clear(); // Важно очищать контекст!
    }
}

String correlationId = UUID.randomUUID().toString();
MDC.put("correlationId", correlationId);
log.info("Отправка синхронного запроса");
 
record.headers().add(new RecordHeader("X-Correlation-ID", correlationId.getBytes()));
ResponseDto response = replyingTemplate.sendAndReceive(record).get();
 
log.info("Получен ответ: {}", response);

// Создание и настройка Circuit Breaker
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50) // Порог ошибок в процентах
    .waitDurationInOpenState(Duration.ofSeconds(30)) // Время в открытом состоянии
    .permittedNumberOfCallsInHalfOpenState(10) // Пробные вызовы в полуоткрытом состоянии
    .slidingWindowType(SlidingWindowType.COUNT_BASED)
    .slidingWindowSize(100) // Размер окна для расчета % ошибок
    .build();
 
CircuitBreaker circuitBreaker = CircuitBreaker.of("kafka-reply", config);
 
// Применение к синхронному запросу
public ResponseDto sendWithProtection(RequestDto request) {
    return circuitBreaker.executeSupplier(() -> {
        try {
            return replyingKafkaTemplate.sendAndReceive(
                new ProducerRecord<>("request-topic", request))
                .get(5, TimeUnit.SECONDS)
                .value();
        } catch (Exception e) {
            throw new RuntimeException("Ошибка синхронного запроса", e);
        }
    });
}

BulkheadConfig bulkheadConfig = BulkheadConfig.custom()
    .maxConcurrentCalls(20)
    .maxWaitDuration(Duration.ofMillis(500))
    .build();
Bulkhead bulkhead = Bulkhead.of("kafka-requests", bulkheadConfig);
 
// Композиция паттернов
Supplier<ResponseDto> decoratedSupplier = Bulkhead.decorateSupplier(
    bulkhead, 
    CircuitBreaker.decorateSupplier(
        circuitBreaker, 
        () -> sendSynchronousRequest(request)
    )
);

// Публикация события изменения
OrderCreatedEvent event = new OrderCreatedEvent(orderId, items, amount);
kafkaTemplate.send("order-events", orderId, event);
 
// Восстановление состояния из потока событий
KStream<String, Event> events = streamsBuilder.stream("order-events");
KTable<String, Order> orderState = events
    .groupByKey()
    .aggregate(
        Order::new,
        (key, event, order) -> order.apply(event)
    );

// Команда - синхронная обработка
CommandResult result = commandService.processCommand(createOrderCommand);
if (result.isSuccess()) {
    // Показать пользователю подтверждение
}
 
// Проекция - асинхронное обновление
@KafkaListener(topics = "order-events")
public void updateReadModel(OrderEvent event) {
    // Обновляем денормализованное представление для быстрого чтения
    readModelRepository.updateOrderProjection(event);
}

// Клиент получает ID события и использует его при чтении
CommandResult result = commandService.processCommand(command);
String eventId = result.getEventId();
 
// При запросе данных указываем, что проекция должна 
// включать события вплоть до указанного
OrderDto order = queryService.getOrder(orderId, eventId);

@SpringBootTest
@Testcontainers
class KafkaRequestReplyIntegrationTest {
 
@Container
static KafkaContainer kafkaContainer = new KafkaContainer(
    DockerImageName.parse("confluentinc/cp-kafka:7.3.0"));
 
@DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
    registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
}
 
@Test
void testRequestReply() {
    // Тесты запрос-ответ с реальным Kafka
}
}

@Test
void testAsyncProcessing() {
    kafkaTemplate.send("test-topic", "test-message");
    
    await().atMost(5, TimeUnit.SECONDS)
        .untilAsserted(() -> 
            verify(messageHandler).handleMessage(argThat(m -> 
                m.getPayload().equals("test-message"))));
}

@Bean
public NewTopic requestTopic() {
    return TopicBuilder.name("request-topic")
        .partitions(3)
        .replicas(1)
        .build();
}

kafka:
  volumes:
    - kafka-data:/var/lib/kafka/data
volumes:
  kafka-data:
    driver: local

src/
├── main/
│   ├── java/com/example/kafkademo/
│   │   ├── config/
│   │   │   ├── KafkaConfig.java
│   │   │   └── TopicConfig.java
│   │   ├── controller/
│   │   │   └── OrderController.java
│   │   ├── model/
│   │   │   ├── Order.java
│   │   │   ├── OrderStatus.java
│   │   │   └── PaymentResult.java
│   │   ├── service/
│   │   │   ├── OrderService.java
│   │   │   ├── PaymentService.java
│   │   │   └── NotificationService.java
│   │   └── KafkaDemoApplication.java
│   └── resources/
│       └── application.yml

public class Order {
    private String id;
    private String customerId;
    private BigDecimal amount;
    private List<OrderItem> items;
    private OrderStatus status;
    
    // Геттеры, сеттеры, конструкторы
}
 
public enum OrderStatus {
    CREATED, PAID, PROCESSING, SHIPPED, DELIVERED, CANCELLED
}
 
public class PaymentResult {
    private String orderId;
    private boolean success;
    private String transactionId;
    private String errorMessage;
    
    // Геттеры, сеттеры, конструкторы
}

@Configuration
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    
    // Общие конфигурации для продюсера и консьюмера
    
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
    
    @Bean
    public ReplyingKafkaTemplate<String, Order, PaymentResult> replyingKafkaTemplate(
            ProducerFactory<String, Order> pf,
            KafkaMessageListenerContainer<String, PaymentResult> container) {
        ReplyingKafkaTemplate<String, Order, PaymentResult> template = 
            new ReplyingKafkaTemplate<>(pf, container);
        template.setDefaultReplyTimeout(Duration.ofSeconds(10));
        return template;
    }
    
    @Bean
    public KafkaMessageListenerContainer<String, PaymentResult> replyContainer(
            ConsumerFactory<String, PaymentResult> cf) {
        ContainerProperties props = new ContainerProperties("payment-results");
        return new KafkaMessageListenerContainer<>(cf, props);
    }
}

@Service
public class PaymentService {
    @KafkaListener(topics = "payment-requests")
    @SendTo
    public PaymentResult processPayment(Order order) {
        log.info("Обработка платежа для заказа {}", order.getId());
        
        // Имитация обработки платежа
        boolean success = Math.random() > 0.2; // 80% успешных платежей
        
        PaymentResult result = new PaymentResult();
        result.setOrderId(order.getId());
        result.setSuccess(success);
        
        if (success) {
            result.setTransactionId(UUID.randomUUID().toString());
        } else {
            result.setErrorMessage("Недостаточно средств");
        }
        
        return result; // Автоматически отправится в топик payment-results
    }
}

@Service
public class NotificationService {
    @KafkaListener(topics = "order-updates")
    public void sendNotification(Order order) {
        log.info("Отправка уведомления для заказа {}: статус={}",
                order.getId(), order.getStatus());
        
        // Имитация отправки email или push-уведомления
        log.info("Уведомление отправлено");
    }
}

@Service
public class OrderService {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final ReplyingKafkaTemplate<String, Order, PaymentResult> replyingTemplate;
    
    // Конструктор с внедрением зависимостей
    
    public Order createOrder(Order order) {
        order.setId(UUID.randomUUID().toString());
        order.setStatus(OrderStatus.CREATED);
        
        // Асинхронная публикация события создания заказа
        kafkaTemplate.send("order-updates", order.getId(), order);
        
        return order;
    }
    
    public PaymentResult processPayment(Order order) throws Exception {
        // Синхронный запрос на обработку платежа
        ProducerRecord<String, Order> record = 
            new ProducerRecord<>("payment-requests", order.getId(), order);
        
        RequestReplyFuture<String, Order, PaymentResult> future = 
            replyingTemplate.sendAndReceive(record);
        
        PaymentResult result = future.get(15, TimeUnit.SECONDS);
        
        // Обновляем статус заказа
        if (result.isSuccess()) {
            order.setStatus(OrderStatus.PAID);
        } else {
            order.setStatus(OrderStatus.CANCELLED);
        }
        
        // Асинхронно публикуем обновление
        kafkaTemplate.send("order-updates", order.getId(), order);
        
        return result;
    }
}

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    private final OrderService orderService;
    
    // Конструктор с внедрением зависимостей
    
    @PostMapping
    public ResponseEntity<Order> createOrder(@RequestBody Order order) {
        Order created = orderService.createOrder(order);
        return ResponseEntity.ok(created);
    }
    
    @PostMapping("/{id}/payment")
    public ResponseEntity<PaymentResult> processPayment(@PathVariable String id) {
        try {
            Order order = orderService.findById(id);
            if (order == null) {
                return ResponseEntity.notFound().build();
            }
            
            PaymentResult result = orderService.processPayment(order);
            return ResponseEntity.ok(result);
        } catch (TimeoutException e) {
            return ResponseEntity.status(HttpStatus.GATEWAY_TIMEOUT)
                .body(new PaymentResult(id, false, null, "Таймаут обработки платежа"));
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(new PaymentResult(id, false, null, e.getMessage()));
        }
    }
    
    @GetMapping("/{id}/track")
    public SseEmitter trackOrder(@PathVariable String id) {
        SseEmitter emitter = new SseEmitter(300000L); // 5 минут
        orderService.subscribeToUpdates(id, order -> {
            try {
                emitter.send(order);
                if (order.getStatus() == OrderStatus.DELIVERED || 
                    order.getStatus() == OrderStatus.CANCELLED) {
                    emitter.complete();
                }
            } catch (IOException e) {
                emitter.completeWithError(e);
            }
        });
        return emitter;
    }
}

@Service
public class OrderService {
    // ... предыдущий код ...
    
    private final Map<String, List<Consumer<Order>>> orderSubscribers = new ConcurrentHashMap<>();
    
    @KafkaListener(topics = "order-updates")
    public void handleOrderUpdate(Order order) {
        log.info("Получено обновление заказа: {}", order);
        
        // Уведомляем всех подписчиков
        List<Consumer<Order>> subscribers = orderSubscribers.get(order.getId());
        if (subscribers != null) {
            subscribers.forEach(subscriber -> subscriber.accept(order));
        }
    }
    
    public void subscribeToUpdates(String orderId, Consumer<Order> subscriber) {
        orderSubscribers.computeIfAbsent(orderId, k -> new CopyOnWriteArrayList<>())
            .add(subscriber);
    }
}

@Service
public class OrderProcessingService {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    // Конструктор с внедрением зависимостей
    
    @KafkaListener(topics = "order-updates")
    public void processOrderUpdates(Order order) {
        if (order.getStatus() == OrderStatus.PAID) {
            // Симуляция логистического процесса
            scheduler.schedule(() -> moveToNextStage(order, OrderStatus.PROCESSING), 
                5, TimeUnit.SECONDS);
        } else if (order.getStatus() == OrderStatus.PROCESSING) {
            scheduler.schedule(() -> moveToNextStage(order, OrderStatus.SHIPPED), 
                10, TimeUnit.SECONDS);
        } else if (order.getStatus() == OrderStatus.SHIPPED) {
            scheduler.schedule(() -> moveToNextStage(order, OrderStatus.DELIVERED), 
                15, TimeUnit.SECONDS);
        }
    }
    
    private void moveToNextStage(Order order, OrderStatus nextStatus) {
        order.setStatus(nextStatus);
        kafkaTemplate.send("order-updates", order.getId(), order);
    }
}