«Настройка логики повтора в Spring Batch»

«1. Обзор

По умолчанию пакетное задание Spring завершается сбоем из-за любых ошибок, возникших во время его выполнения. Однако иногда нам может понадобиться повысить устойчивость нашего приложения, чтобы справляться с периодическими сбоями.

В этом кратком руководстве мы рассмотрим, как настроить логику повторных попыток в среде Spring Batch.

2. Пример использования

Допустим, у нас есть пакетное задание, которое читает входной CSV-файл:

username, userid, transaction_date, transaction_amount
sammy, 1234, 31/10/2015, 10000
john, 9999, 3/12/2015, 12321

Затем оно обрабатывает каждую запись, обращаясь к конечной точке REST, чтобы получить возраст пользователя и почтовый индекс. атрибуты:

public class RetryItemProcessor implements ItemProcessor<Transaction, Transaction> {
    
    @Override
    public Transaction process(Transaction transaction) throws IOException {
        log.info("RetryItemProcessor, attempting to process: {}", transaction);
        HttpResponse response = fetchMoreUserDetails(transaction.getUserId());
        //parse user's age and postCode from response and update transaction
        ...
        return transaction;
    }
    ...
}

И, наконец, он генерирует объединенный выходной XML:

<transactionRecord>
    <transactionRecord>
        <amount>10000.0</amount>
        <transactionDate>2015-10-31 00:00:00</transactionDate>
        <userId>1234</userId>
        <username>sammy</username>
        <age>10</age>
        <postCode>430222</postCode>
    </transactionRecord>
    ...
</transactionRecord>

3. Добавление повторов в ItemProcessor

А что, если время ожидания подключения к конечной точке REST истекло из-за некоторой медлительности сети ? Если это так, наше пакетное задание завершится ошибкой.

В таких случаях мы бы предпочли, чтобы неудачная обработка элемента была повторена пару раз. Итак, давайте настроим наше пакетное задание на выполнение до трех повторных попыток в случае сбоя:

@Bean
public Step retryStep(
  ItemProcessor<Transaction, Transaction> processor,
  ItemWriter<Transaction> writer) throws ParseException {
    return stepBuilderFactory
      .get("retryStep")
      .<Transaction, Transaction>chunk(10)
      .reader(itemReader(inputCsv))
      .processor(processor)
      .writer(writer)
      .faultTolerant()
      .retryLimit(3)
      .retry(ConnectTimeoutException.class)
      .retry(DeadlockLoserDataAccessException.class)
      .build();
}

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

4. Тестирование повторов

Давайте представим тестовый сценарий, в котором конечная точка REST, возвращающая возраст и почтовый код, не работала какое-то время. В этом тестовом сценарии мы получим исключение ConnectTimeoutException только для первых двух вызовов API, а третий вызов завершится успешно:

@Test
public void whenEndpointFailsTwicePasses3rdTime_thenSuccess() throws Exception {
    FileSystemResource expectedResult = new FileSystemResource(EXPECTED_OUTPUT);
    FileSystemResource actualResult = new FileSystemResource(TEST_OUTPUT);

    when(httpResponse.getEntity())
      .thenReturn(new StringEntity("{ \"age\":10, \"postCode\":\"430222\" }"));
 
    //fails for first two calls and passes third time onwards
    when(httpClient.execute(any()))
      .thenThrow(new ConnectTimeoutException("Timeout count 1"))
      .thenThrow(new ConnectTimeoutException("Timeout count 2"))
      .thenReturn(httpResponse);

    JobExecution jobExecution = jobLauncherTestUtils
      .launchJob(defaultJobParameters());
    JobInstance actualJobInstance = jobExecution.getJobInstance();
    ExitStatus actualJobExitStatus = jobExecution.getExitStatus();

    assertThat(actualJobInstance.getJobName(), is("retryBatchJob"));
    assertThat(actualJobExitStatus.getExitCode(), is("COMPLETED"));
    AssertFile.assertFileEquals(expectedResult, actualResult);
}

Здесь наша работа выполнена успешно. Кроме того, из журналов видно, что первая запись с id=1234 дважды не удалась, а третья попытка завершилась успешно:

19:06:57.742 [main] INFO  o.s.batch.core.job.SimpleStepHandler - Executing step: [retryStep]
19:06:57.758 [main] INFO  o.b.batch.service.RetryItemProcessor - Attempting to process user with id=1234
19:06:57.758 [main] INFO  o.b.batch.service.RetryItemProcessor - Attempting to process user with id=1234
19:06:57.758 [main] INFO  o.b.batch.service.RetryItemProcessor - Attempting to process user with id=1234
19:06:57.758 [main] INFO  o.b.batch.service.RetryItemProcessor - Attempting to process user with id=9999
19:06:57.773 [main] INFO  o.s.batch.core.step.AbstractStep - Step: [retryStep] executed in 31ms

Точно так же давайте рассмотрим еще один тестовый пример, чтобы увидеть, что происходит, когда все попытки исчерпаны: ~~ ~

@Test
public void whenEndpointAlwaysFail_thenJobFails() throws Exception {
    when(httpClient.execute(any()))
      .thenThrow(new ConnectTimeoutException("Endpoint is down"));

    JobExecution jobExecution = jobLauncherTestUtils
      .launchJob(defaultJobParameters());
    JobInstance actualJobInstance = jobExecution.getJobInstance();
    ExitStatus actualJobExitStatus = jobExecution.getExitStatus();

    assertThat(actualJobInstance.getJobName(), is("retryBatchJob"));
    assertThat(actualJobExitStatus.getExitCode(), is("FAILED"));
    assertThat(actualJobExitStatus.getExitDescription(),
      containsString("org.apache.http.conn.ConnectTimeoutException"));
}

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

5. Настройка повторов с использованием XML

Наконец, давайте посмотрим на XML-эквивалент приведенных выше конфигураций:

<batch:job id="retryBatchJob">
    <batch:step id="retryStep">
        <batch:tasklet>
            <batch:chunk reader="itemReader" writer="itemWriter"
              processor="retryItemProcessor" commit-interval="10"
              retry-limit="3">
                <batch:retryable-exception-classes>
                    <batch:include class="org.apache.http.conn.ConnectTimeoutException"/>
                    <batch:include class="org.springframework.dao.DeadlockLoserDataAccessException"/>
                </batch:retryable-exception-classes>
            </batch:chunk>
        </batch:tasklet>
    </batch:step>
</batch:job>

6. Заключение

В этой статье мы узнали, как настроить логику повторов в Spring Партия. Мы рассмотрели конфигурации Java и XML.

Мы также использовали модульный тест, чтобы увидеть, как повторные попытки работают на практике.

Как всегда, пример кода для этого руководства доступен на GitHub.