Блог

Я использую Spring batch для написания пакетного процесса, и у меня возникают проблемы с обработкой исключений.

У меня есть читатель, который извлекает элементы из базы данных с определенным состоянием. Считыватель передает элемент на шаг процессора, который может запустить исключение MyException.class . Когда возникает это исключение, я хочу пропустить элемент, вызвавший это исключение, и продолжить чтение следующего.

Проблема здесь в том, что мне нужно изменить состояние этого элемента в базе данных, чтобы он не был снова извлечен читателем.

Это то, что я пытался:

 return this.stepBuilderFactory.get("name")
            .<Input, Output>chunk(1)
            .reader(reader())
            .processor(processor())
            .faultTolerant()
            .skipPolicy(skipPolicy())
            .writer(writer())
            .build();
 

В моем классе SkipPolicy у меня есть следующий код:

 public boolean shouldSkip(Throwable throwable, int skipCount) throws SkipLimitExceededException {
    if (throwable instanceof MyException.class) {
        // log the issue
        // update the item that caused the exception in database so the reader doesn't return it again
        return true;
    }
    return false;
}
 

С помощью этого кода исключение пропускается, и мой читатель вызывается снова, однако политика пропусков не зафиксировала изменения или не выполнила откат, поэтому читатель извлекает элемент и пытается обработать его снова.

Я также попытался сделать исключение из правил:

 return this.stepBuilderFactory.get("name")
            .<Input, Output>chunk(1)
            .reader(reader())
            .processor(processor())
            .faultTolerant()
            .skip(MyException.class)
            .exceptionHandler(myExceptionHandler())
            .writer(writer())
            .build();
 

В моем классе ExceptionHandler у меня есть следующий код:

 public void handleException(RepeatContext context, Throwable throwable) throws Throwable {
     if (throwable.getCause() instanceof MyException.class) {
        // log the issue
        // update the item that caused the exception in database so the reader doesn't return it again
     } else {
        throw throwable;
     }
}
 

С помощью этого решения состояние изменяется в базе данных, однако оно не вызывает считыватель, вместо этого оно process снова вызывает метод processor() , попадая в бесконечный цикл.

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

Что я делаю не так?

РЕДАКТИРОВАТЬ: После множества тестов и использования разных прослушивателей , например SkipListener , я не смог добиться того, чего хотел, Spring Batch всегда выполняет откат моего ОБНОВЛЕНИЯ.

Отладка вот что я нашел:

Как только мой прослушиватель вызывается и я обновляю свой элемент, программа вводит метод write в класс FaultTolerantChunkProcessor (строка #327). Этот метод попытается выполнить следующий код (скопированный с github):

 try {
    doWrite(outputs.getItems());
} catch (Exception e) {
    status = BatchMetrics.STATUS_FAILURE;
    if (rollbackClassifier.classify(e)) {
        throw e;
    }
    /*
     * If the exception is marked as no-rollback, we need to
     * override that, otherwise there's no way to write the
     * rest of the chunk or to honour the skip listener
     * contract.
     */
     throw new ForceRollbackForWriteSkipException(
        "Force rollback on skippable exception so that skipped item can be located.", e);
}
 

Метод doWrite (строка №151) внутри класса SimpleChunkProcessor попытается записать список выходных элементов, однако в моем случае список пуст, поэтому в строке № 159 (метод writeItems ) запустится IndexOutOfBoundException , вызывая ForceRollbackForWriteSkipException и выполняя откат, от которого я страдаю.

Если я переопределяю класс FaultTolerantChunkProcessor и избегаю записи элементов, если список пуст, то все работает так, как задумано, обновление выполняется, и программа пропускает ошибку и снова вызывает считыватель.

Я не знаю, действительно ли это ошибка или она вызвана чем-то, что я делаю неправильно в своем коде.

SkipListener На мой взгляд, A лучше подходит для вашего варианта использования, чем an ExceptionHandler , так как он дает вам доступ к элементу, вызвавшему исключение. С помощью обработчика исключений вам необходимо перенести элемент в контекст исключения или повторения.

Кроме того, прослушиватель пропуска позволяет узнать, на какой фазе произошло исключение (т. е. при чтении, обработке или записи), в то время как с обработчиком исключений вам нужно найти способ обнаружить это самостоятельно. Если код пропуска одинаков для всех этапов, вы можете вызвать один и тот же метод, который обновляет статус элемента во всех методах прослушивателя.