Как обрабатывать ошибки в Kafka Consumer

Question

У меня есть следующий класс конфигурации Kafka:

@Configuration
@AllArgsConstructor(access = AccessLevel.PROTECTED)

public class KafkaConfiguration {
private final KafkaConfigurationProperties kafkaConfigurationProperties;

@Bean
public ConcurrentKafkaListenerContainerFactory<String, RepaymentEvent> debtCollectorConsumerContainerFactory() {
     ConcurrentKafkaListenerContainerFactory<String, RepaymentEvent> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfiguration()));
    factory.setConcurrency(kafkaConfigurationProperties.getDebtCollectorConsumerThreads());
    factory.setStatefulRetry(true);
    factory.setErrorHandler(new SeekToCurrentErrorHandler((record, exception) -> {
        if (exception instanceof SomeCustomException) {
            // here I want to mannually Acknowledge the consuming of the record
        }
    }, 10));

    ContainerProperties containerProperties = factory.getContainerProperties();
    containerProperties.setAckOnError(false);
    containerProperties.setAckMode(ContainerProperties.AckMode.RECORD);
    return factory;
}

@Bean
@Qualifier(KAFKA_LOAN_REPAYMENT_PRODUCER)
public Producer<String, RepaymentEvent> loanRepaymentProducer() {
    return new KafkaProducer<>(producerConfiguration());
}

@Bean
@Qualifier(KAFKA_DEBT_COLLECTOR_PRODUCER)
public Producer<String, RepaymentEvent> debtCollectorProducer() {
    return new KafkaProducer<>(producerConfiguration());
}

private Map<String, Object> consumerConfiguration() {
    Map<String, Object> properties = new HashMap<>();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfigurationProperties.getBootstrapServers());
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConfigurationProperties.getDebtCollectorConsumerGroupId());
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConfigurationProperties.getDebtCollectorConsumerAutoOffsetReset());
    properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConfigurationProperties.getDebtCollectorConsumerMaxPollRecords());
    properties.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, Boolean.TRUE);
    properties.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, kafkaConfigurationProperties.getConfluentEndpoint());
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.FALSE);
    return properties;
}

private Map<String, Object> producerConfiguration() {
    Map<String, Object> properties = new HashMap<>();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfigurationProperties.getBootstrapServers());
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
    properties.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, kafkaConfigurationProperties.getConfluentEndpoint());
    return properties;
}
}

и следующий KafkaListener:

@Slf4j
@Component
@AllArgsConstructor(access = AccessLevel.PROTECTED)
public class DebtCollectorIncomingClient {

private final RepaymentTransferProcessService repaymentTransferProcessService;

@KafkaListener(
        topics = "${kafka.debtCollectorIncomingTopic}",
        groupId = "${kafka.debtCollectorConsumerAutoOffsetReset}",
        containerFactory = "debtCollectorConsumerContainerFactory")
public void submitMoneyTransferCommand(@Payload RepaymentEvent repaymentEvent) {
    log.info("Receiving command: {}", repaymentEvent);
    if (repaymentEvent.getPayload() instanceof RepaymentRequestTransfer) {
        RepaymentTransfer repaymentTransfer = aRepaymentTransfer(repaymentEvent);
        repaymentTransferProcessService.startRepaymentTransferProcess(repaymentTransfer);
    }
}

private RepaymentTransfer aRepaymentTransfer(RepaymentEvent repaymentEvent) {
    RepaymentRequestTransfer repaymentRequestTransfer = (RepaymentRequestTransfer) repaymentEvent.getPayload();
    return RepaymentTransfer.builder()
            .clientId(repaymentRequestTransfer.getClientId())
            .contractId(repaymentRequestTransfer.getContractId())
            .amount(BigDecimal.valueOf(repaymentRequestTransfer.getAmount()))
            .currency(Currency.getInstance(repaymentRequestTransfer.getCurrency().name()))
            .debtCollectorExternalId(repaymentEvent.getCorrelationId())
            .debtType(repaymentRequestTransfer.getDebtType())
            .build();
}
}

Я хочу использовать SeekToCurrentErrorHandler для обработки ошибок и хочу что-то конкретное, например здесь , но в настоящее время я использую springBootVersion=2.0.4.RELEASE, springKafkaVersion=2.1.4.RELEASE, kafkaVersion=2.0.1 и confluentVersion=3.3.1. Можете ли вы помочь мне установить зависимости и конфигурацию для обработки ошибок в Kafka?

Привет!

Answer 1

SeekToCurrentErrorHandler доступен с версии 2.0.1. Дополнительный функционал (восстановление после некоторого количества попыток) был добавлен в версии 2.2.

Используйте Spring Boot 2.1.4 и Spring для Apache Kafka 2.2.6 (скоро будет доступна загрузка 2.1.5).

Answer 2

Через несколько дней и прочитав ответы Гэри на некоторых других постах, я наконец нашел решение своей проблемы.Может быть, вопрос не очень описательный, но этот ответ описывает поведение, которое я хочу.

в @Configuration Я создаю следующий компонент Spring:

@Bean
    public ConcurrentKafkaListenerContainerFactory<String, RepaymentEvent> debtCollectorConsumerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, RepaymentEvent> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfiguration()));
        factory.setConcurrency(kafkaConfigurationProperties.getDebtCollectorConsumerThreads());
        factory.setErrorHandler(new BlockingSeekToCurrentErrorHandler());

        ContainerProperties containerProperties = factory.getContainerProperties();
        containerProperties.setAckOnError(false);
        containerProperties.setAckMode(ContainerProperties.AckMode.RECORD);

        factory.setRetryTemplate(retryTemplate());
        return factory;
    }

private RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setBackOffPolicy(backOffPolicy());
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(kafkaConfigurationProperties.getDebtCollectorConsumerRetryAttempts()));
    return retryTemplate;
}

И BlockingSeekToCurrentErrorHandler класс:

public class BlockingSeekToCurrentErrorHandler extends SeekToCurrentErrorHandler {

    private static final int MAX_RETRY_ATTEMPTS = Integer.MAX_VALUE;

    BlockingSeekToCurrentErrorHandler() {
        super(MAX_RETRY_ATTEMPTS);
    }

    @Override
    public void handle(Exception exception, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
        try {
            if (!records.isEmpty()) {
                log.warn("Exception: {} occurred with message: {}", exception, exception.getMessage());
                MetricFactory.handleDebtCollectorIncomingBlockingError(records.get(0), exception);
                super.handle(exception, records, consumer, container);
            }
        } catch (SerializationException e) {
            log.warn("Exception: {} occurred with message: {}", e, e.getMessage());
            MetricFactory.handleDebtCollectorIncomingDeserializationError(records, e);
        }
    }
}