如何在Spring Batch RetryContext中设置值

huangapple
go评论102阅读模式
英文:

How to set values in Spring Batch RetryContext

问题

我有一个用于批处理作业的步骤定义如下：

  2.   @Autowired
  3.   RetryContextCache retryContextCache;
  5.   @Bean
  6.   public Step correctionStep(JpaTransactionManager transactionManager) {
  8.     return new StepBuilder("correction-step", jobRepository)
  9.         .<String, List<BookingInfo>>chunk(10, transactionManager)
  10.         .reader(customItemReader())
  11.         .processor(correctionProcessor())
  12.         .writer(customItemWriter)
  13.         .taskExecutor(correctionTaskExecutor())
  14.         .faultTolerant()
  15.         .retryPolicy(retryPolicy())
  16.         .backOffPolicy(exponentialBackOffPolicy())
  17.         .retryContextCache(retryContextCache)
  18.         .skipPolicy(skipPolicy())
  19.         .build();
  20.   }

我有RetryContext和RetryContextCache的bean定义：

  1.   @Bean
  2.   RetryContextCache retryContextCache() {
  3.     return new MapRetryContextCache();
  4.   }
  6.   @Bean
  7.   RetryContext retryContext() {
  8.     return new RetryContextSupport(null);
  9.   }

现在，我在processor中使用它们，如下所示：

  1. @Component
  2. public class CorrectionProcessor implements ItemProcessor <String, List <BookingInfo>> {
  4.   @Autowired
  5.   RetryContextCache retryContextCache;
  7.   @Autowired
  8.   RetryContext retryContext;
  10.   public List<BookingInfo> process(String bookingId) throws Exception {
  12.     List<BookingInfo> list = new ArrayList <> ();
  14.     if (retryContextCache.containsKey(bookingId)) {
  15.       list = (List<BookingInfo>) retryContextCache.get(bookingId).getAttribute(bookingId);
  16.     } else {
  17.       // fetch and populate list from database.
  18.     }
  20.     try {
  21.       // do something with the list.
  22.     } catch (Exception e) {
  23.       // modify something in the list.
  24.       retryContext.setAttribute(bookingId, list);
  25.       retryContextCache.put(bookingId, retryContext);
  26.       throw e;
  27.     }
  29.   }
  30. }

你可以看到，在重新抛出异常以便重试机制生效之前，我尝试在retryContextCache中设置一些值。

当重试发生时，它会进入上面代码中提到的if条件。但是，retryContextCache.get(bookingId).getAttribute(bookingId)的值始终为null。

我是否错误地设置了重试上下文中的值？为什么不起作用？

英文:

I have the below step definition for a batch job:

  2.   @Autowired
  3.   RetryContextCache retryContextCache;
  5.   @Bean
  6.   public Step correctionStep(JpaTransactionManager transactionManager) {
  8.     return new StepBuilder("correction-step", jobRepository)
  9.         .<String, List<BookingInfo>>chunk(10, transactionManager)
  10.         .reader(customItemReader())
  11.         .processor(correctionProcessor())
  12.         .writer(customItemWriter)
  13.         .taskExecutor(correctionTaskExecutor())
  14.         .faultTolerant()
  15.         .retryPolicy(retryPolicy())
  16.         .backOffPolicy(exponentialBackOffPolicy())
  17.         .retryContextCache(retryContextCache)
  18.         .skipPolicy(skipPolicy())
  19.         .build();
  20.   }

I have bean definitions for RetryContext and RetryContextCache:

  1.   @Bean
  2.   RetryContextCache retryContextCache() {
  3.     return new MapRetryContextCache();
  4.   }
  6.   @Bean
  7.   RetryContext retryContext() {
  8.     return new RetryContextSupport(null);
  9.   }

Now, I use them in the processor like below:

  1. @Component
  2. public class CorrectionProcessor implements ItemProcessor <String, List <BookingInfo>> {
  4.   @Autowired
  5.   RetryContextCache retryContextCache;
  7.   @Autowired
  8.   RetryContext retryContext;
  10.   public List<BookingInfo> process(String bookingId) throws Exception {
  12.     List<BookingInfo> list = new ArrayList <> ();
  14.     if (retryContextCache.containsKey(bookingId)) {
  15.       list = (List<BookingInfo>) retryContextCache.get(bookingId).getAttribute(bookingId);
  16.     } else {
  17.       // fetch and populate list from database.
  18.     }
  20.     try {
  21.       // do something with the list.
  22.     } catch (Exception e) {
  23.       // modify something in the list.
  24.       retryContext.setAttribute(bookingId, list);
  25.       retryContextCache.put(bookingId, retryContext);
  26.       throw e;
  27.     }
  29.   }
  30. }

You can see that I try to set some values in the retryContextCache before I re-throw the exception for the retry mechanism to work.

When retry happens, it goes inside the if condition mentioned in the above code.
But, the value of retryContextCache.get(bookingId).getAttribute(bookingId) is always null.

Am I setting the value in retry context incorrectly? Why is this not working?

答案1

得分: 0

这是我解决问题的方法。

我想在重试开始之前保存对象状态。RetryContext 的作用范围从错误抛出后立即开始。所以，我无法在 RetryContext 中设置值。

因此，我创建了一个带有 StepScope 的 bean：

  1. @Bean
  2. @StepScope
  3. public Map<String, BookingInfo> objectStateMap() {
  4.  return new ConcurrentHashMap<>();
  5. }

然后，我在需要的地方自动装配了这个哈希表。

接着，在捕获块中的错误重新抛出之前，我将修改后的对象写入了这个哈希表。

然后，我为我的用例创建了一个 SkipListener。这将在失败时捕获被跳过的对象。然后，它将根据需要执行它们各自的任务。

  1. @Component
  2. @Slf4j
  3. public class CustomSkipListener {
  5.   @Autowired
  6.   Map<String, List<BookingInfo>> objectStateMap;
  9.   @OnSkipInRead
  10.   public void onSkipInRead(Throwable t) {
  11.     log.error("Read has skipped because of error : " + t.getMessage());
  12.   }
  14.   @OnSkipInWrite
  15.   public void onSkipInWrite(List<BookingInfo> item, Throwable t) {
  16.     log.error("Write has skipped because of error : " + t.getMessage());
  17.   }
  19.   @OnSkipInProcess
  20.   public void onSkipInProcess(String bookingId, Throwable t) {
  21.     List<BookingInfo> bookingInfos = objectStateMap.get(bookingId);
  22.     // do some tasks..
  23.     objectStateMap.remove(bookingId);
  24.   }
  26. }

只需将这个 listener 注册到主要的作业中。

英文:

Here is how I resolved the issue.

I wanted to save object state before the retry starts. The scope for RetryContext starts right after the error is thrown. So, I was not able to set the value in RetryContext.

So, I created a bean with StepScope:

  1. @Bean
  2. @StepScope
  3. public Map&lt;String,BookingInfo&gt; objectStateMap() {
  4.  return new ConcurrentHashMap&lt;&gt;();
  5. }

And then, I autowired this hash wherever required.

Then, I wrote the modified objects to this hash before the error is re-thrown in the catch block.

Then, I created a SkipListener for my usecase. This will capture the skipped object in case of failure. And then, it will do their respective task accordingly.

  1. @Component
  2. @Slf4j
  3. public class CustomSkipListener {
  5.   @Autowired
  6.   Map&lt;String, List&lt;BookingInfo&gt;&gt; objectStateMap;
  9.   @OnSkipInRead
  10.   public void onSkipInRead(Throwable t) {
  11.     log.error(&quot;Read has skipped because of error : &quot; + t.getMessage());
  12.   }
  14.   @OnSkipInWrite
  15.   public void onSkipInWrite(List&lt;BookingInfo&gt; item, Throwable t) {
  16.     log.error(&quot;Write has skipped because of error : &quot; + t.getMessage());
  17.   }
  19.   @OnSkipInProcess
  20.   public void onSkipInProcess(String bookingId, Throwable t) {
  21.     List&lt;BookingInfo&gt; bookingInfos = objectStateMap.get(bookingId);
  22.     // do some tasks..
  23.     objectStateMap.remove(bookingId);
  24.   }
  26. }

Just registered this listener into the main job.

huangapple
  • 本文由 发表于 2023年6月15日 02:07:58
  • 转载请务必保留本文链接：https://go.coder-hub.com/76476432.html
  • java
  • spring
  • spring-batch
  • spring-boot
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定