Обработка исключений в Spring Kafka - PullRequest
Новая
       6

Обработка исключений в Spring Kafka

0 голосов
/ 12 февраля 2020

пользуюсь spring-kafka 2.2.6. Я использовал SeekToCurrentErrorHandler и ErrorHandlingDeserializer2. SeekToCurrentErrorHandler в настоящее время настроен на запись сообщений после трех попыток. Есть ли способ пропустить повторы для ошибок проверки (обнаруженных реализацией Validator в Spring) и ошибок преобразования сообщений? Все ошибки перехватываются обработчиком ошибок контейнера, т.е. SeeToCurrentErrorHandler. Должен ли я переопределить метод дескриптора SeeToCurrentErrorHandler? 

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(this.kafkaConfigProperties.getConsumerConcurrency());
    factory.setAutoStartup(false);
    factory.setErrorHandler(new SeekToCurrentErrorHandler((c, e) -> {
        LOG.info(e.getMessage());
    }, this.kafkaConfigProperties.getRetryCount()));
    return factory;
}

 @Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> map = new HashMap<>();
    Properties consumerProperties = getConsumerProperties();
    consumerProperties.forEach((key, value) -> {
        map.put((String) key, value);
    });
    KafkaSoapMessageConverter kafkaSoapMessageConverter = new KafkaSoapMessageConverter();
    Map<String, Object> configMap = new HashMap<>(1);
    configMap.put(KafkaSoapMessageConverter.CLASS_TO_DESERIALIZE, MyClass.class);
    kafkaSoapMessageConverter.configure(configMap, false);
    ErrorHandlingDeserializer2<Object> errorHandlingDeserializer = new ErrorHandlingDeserializer2<>(
            kafkaSoapMessageConverter);
    DefaultKafkaConsumerFactory<String, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(map);
    consumerFactory.setValueDeserializer(errorHandlingDeserializer);
    return consumerFactory;
}

EDIT

Я использовал приведенный ниже код 

if(DeserializationException.class == e.getClass() 
        || e.getCause().getClass() == MethodArgumentNotValidException.class) {
    SeekUtils.doSeeks(records, consumer, e, true, (c, e) -> { return true; }, LOG); 
} else {
    super.handle(e, records, consumer, container);
}

1 Ответ

0 голосов
/ 12 февраля 2020

Версия 2.3 (текущая версия 2.3.5) добавлена ​​возможность настройки, какие исключения повторяются: 

/**
 * Set an exception classifications to determine whether the exception should cause a retry
 * (until exhaustion) or not. If not, we go straight to the recoverer. By default,
 * the following exceptions will not be retried:
 * <ul>
 * <li>{@link DeserializationException}</li>
 * <li>{@link MessageConversionException}</li>
 * <li>{@link MethodArgumentResolutionException}</li>
 * <li>{@link NoSuchMethodException}</li>
 * <li>{@link ClassCastException}</li>
 * </ul>
 * All others will be retried.
 * When calling this method, the defaults will not be applied.
 * @param classifications the classifications.
 * @param defaultValue whether or not to retry non-matching exceptions.
 * @see BinaryExceptionClassifier#BinaryExceptionClassifier(Map, boolean)
 */
public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifications, boolean defaultValue) {

Вот как установлены значения по умолчанию: 

    Map<Class<? extends Throwable>, Boolean> classified = new HashMap<>();
    classified.put(DeserializationException.class, false);
    classified.put(MessageConversionException.class, false);
    classified.put(MethodArgumentResolutionException.class, false);
    classified.put(NoSuchMethodException.class, false);
    classified.put(ClassCastException.class, false);

Также Вы можете добавить исключения к значениям по умолчанию: 

/**
 * Add an exception type to the default list; if and only if an external classifier
 * has not been provided. By default, the following exceptions will not be retried:
 * <ul>
 * <li>{@link DeserializationException}</li>
 * <li>{@link MessageConversionException}</li>
 * <li>{@link MethodArgumentResolutionException}</li>
 * <li>{@link NoSuchMethodException}</li>
 * <li>{@link ClassCastException}</li>
 * </ul>
 * All others will be retried.
 * @param exceptionType the exception type.
 * @see #removeNotRetryableException(Class)
 * @see #setClassifications(Map, boolean)
 */
public void addNotRetryableException(Class<? extends Exception> exceptionType) {
    Assert.isTrue(this.classifier instanceof ExtendedBinaryExceptionClassifier,
            "Cannot add exception types to a supplied classifier");
    ((ExtendedBinaryExceptionClassifier) this.classifier).getClassified().put(exceptionType, false);
}
...