Spring Batch Partitioned Approach: Multiple batches if triggered together-serialize access due to read/write dependencies among transactions error

[PSHREYASHOLLA]

When multiple batches are getting triggered together getting below error, rg.springframework.dao.CannotAcquireLockException: JDBC commit; ERROR: could not serialize access due to read/write dependencies among transactions Detail: Reason code: Canceled on identification as a pivot, during commit attempt. Hint: The transaction might succeed if retried. at org.springframework.jdbc.support.SQLStateSQLExceptionTranslator.doTranslate(SQLStateSQLExceptionTranslator.java:115) at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:70) at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:79) at org.springframework.jdbc.support.JdbcTransactionManager.translateException(JdbcTransactionManager.java:178) at org.springframework.jdbc.datasource.DataSourceTransactionManager.doCommit(DataSourceTransactionManager.java:340) at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:743) at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:711) at org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning(TransactionAspectSupport.java:660) at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:410) at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) at org.springframework.batch.core.repository.support.AbstractJobRepositoryFactoryBean$1.invoke(AbstractJobRepositoryFactoryBean.java:207) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:244) at jdk.proxy2/jdk.proxy2.$Proxy238.createJobExecution(Unknown Source) at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:145) at org.springframework.batch.core.launch.support.TaskExecutorJobLauncher.run(TaskExecutorJobLauncher.java:59) at com.finastra.retailbanking.essence.febp.service.impl.FebpApplicationJobServiceImpl.triggerApplicationBatchJob(FebpApplicationJobServiceImpl.java:165)

@Service
public class FebpApplicationJobServiceImpl implements FebpApplicationJobService {

  /** The LOGGER. */
  private static final Logger LOGGER = LoggerFactory.getLogger(FebpApplicationJobServiceImpl.class);

  /** The app context. */
  @Autowired
  private ApplicationContext appContext;

  /** The job launcher. */
  @Autowired
  @Qualifier("febpBatchJobLauncher")
  private JobLauncher jobLauncher;

  /** The job explorer. */
  @Autowired
  @Qualifier("febpBatchJobExplorer")
  private JobExplorer jobExplorer;

  /** The job repository. */
  @Autowired
  @Qualifier("febpBatchJobRepository")
  private JobRepository jobRepository;

  /** The BatchStatusUpdateThread. */
  private BatchStatusUpdateThread batchStatusUpdateThread;
  /**
   * Trigger or resume application batch job.
   *
   * @param applicationJobDataDTO the application job data DTO
   * @return the response entity
   */
  @Override
  public ResponseEntity<ApplicationJobDataResDTO> triggerOrResumeApplicationBatchJob(ApplicationJobDataReqDTO applicationJobDataDTO) {
    ApplicationJobDataResDTO applicationJobDataResDTO = new ApplicationJobDataResDTO();
    if (applicationJobDataDTO.getAppRestartJobExecutionId() == null || applicationJobDataDTO.getAppRestartJobExecutionId().isEmpty()) {
      BatchStatus appJobStatus = triggerApplicationBatchJob(applicationJobDataDTO, applicationJobDataResDTO);
      applicationJobDataResDTO.setAppJobStatus(appJobStatus != null ? appJobStatus.toString() : null);
    } else {
      BatchStatus appJobResumeStatus = null;
      try {
        appJobResumeStatus = resumeApplicationBatchJob(applicationJobDataDTO.getAppRestartJobExecutionId(), applicationJobDataResDTO);
      } catch (Exception e) {
        LOGGER.error("Error during resume FEBP Batch Job:  ", e);
      }
      applicationJobDataResDTO.setAppJobStatus(appJobResumeStatus != null ? appJobResumeStatus.toString() : null);

    }
    return new ResponseEntity<>(applicationJobDataResDTO, HttpStatus.OK);
  }

  /**
   * Trigger application batch job.
   *
   * @param applicationJobDataDTO    the application job data DTO
   * @param applicationJobDataResDTO
   * @return the batch status
   */
  private BatchStatus triggerApplicationBatchJob(ApplicationJobDataReqDTO applicationJobDataDTO, ApplicationJobDataResDTO applicationJobDataResDTO) {
    JobExecution jobExecution = null;
    try {
      IBatchJobFactory batchJobFactory = (IBatchJobFactory) appContext.getBean(applicationJobDataDTO.getAppJobBeanName());
      Job appJob = batchJobFactory.getBatchJob(ApplicationThreadLocal.getTenantId(), applicationJobDataDTO);
      Map<String, Object> appJobParamsMap = applicationJobDataDTO.getAppJobParamsMap();
      JobParameters appJobParams = configureJobParameters(appJobParamsMap);
      jobExecution = jobLauncher.run(appJob, appJobParams);
      applicationJobDataResDTO.setAppJobExecutionId(String.valueOf(jobExecution.getId()));
      getBatchStatusUpdateThread().addJobToMap(jobExecution.getId(), appJob.getName(), applicationJobDataDTO.getApplicationCode());
      return BatchStatus.STARTED;
    } catch (final JobExecutionAlreadyRunningException e) {
      LOGGER.error("JobExecutionAlreadyRunningException: ", e);
    } catch (final JobRestartException e) {
      LOGGER.error("JobRestartException: ", e);
    } catch (final JobInstanceAlreadyCompleteException e) {
      LOGGER.error("JobInstanceAlreadyCompleteException: ", e);
    } catch (final JobParametersInvalidException e) {
      LOGGER.error("JobParametersInvalidException: ", e);
    } catch (final NoSuchBeanDefinitionException e) {
      LOGGER.error("No Job bean name defined with {}", applicationJobDataDTO.getAppJobBeanName());
      EventMessageSupport.getInstance().raiseBusinessEvent(EventMessageConstant.INVALID_BEAN_NAME, applicationJobDataDTO.getAppJobBeanName());
    } catch (final Exception e) {
      LOGGER.error("Exception: ", e);
    }
    return BatchStatus.FAILED;
  }

  /**
   * Instance of BatchStatusUpdateThread.
   * 
   * @return Instance of batchStatusUpdateThread
   */
  private BatchStatusUpdateThread getBatchStatusUpdateThread() {
    if (batchStatusUpdateThread == null) {
      batchStatusUpdateThread = appContext.getBean(BatchStatusUpdateThread.class);

    }
    return batchStatusUpdateThread;
  } 

And error is at jobExecution = jobLauncher.run(appJob, appJobParams);

[PSHREYASHOLLA]

Hi Team, Can someone help here?

[fmbenhassine]

Hi @PSHREYASHOLLA

I remember working with you on StackOverflow and did my best to answer your questions in details and with complete working examples (answers that you have accepted):

• Restarting a job after server crash is not processing the unprocessed data
• Restarting a FAILED job is not processing the failed chunk data again
• How to make the JobRegistry contain the job information on server restart in spring batch

However, I see you are still struggling with other aspects, which to me, are the same or similar issues that we discussed already. To help you more efficiently, I propose a Zoom session where you can ask your questions live. Is that ok for you? This will be way more efficient than what we have done so far with text on StackOverflow.. Happy to help you move forward on this, because restarting a failed partitioned step should not be that hard (and in my experience, it works pretty well).

[PSHREYASHOLLA]

@fmbenhassine Yes you had helped me earlier with the partitioned approach and the same is working fine. Now here what is happening is, when we get a request to trigger multiple batches[All batches implemented using partitioned approach] simultaneously we are getting this org.springframework.dao.CannotAcquireLockException: JDBC commit; ERROR: could not serialize access due to read/write dependencies among transactions Detail: Reason code: Canceled on identification as a pivot, during commit attempt. Hint: The transaction might succeed if retried. I found one more https://stackoverflow.com/questions/78958090/spring-batch-could-not-serialize-access-due-to-read-write-dependencies-among-t Is this same?

[fmbenhassine]

yes, that's the same. I will add a separate answer here that you can accept if it helps.

[PSHREYASHOLLA]

So here is my code,

@Configuration
public class FEBPDBConfig {

  /** The febp properties. */
  @Autowired
  private FebpProperties febpProperties;

  @Primary
  @Bean(name = "febpDataSource")
  public DataSource batchDataSource() {
    DataSource febpDBSrc = DataSourceBuilder.create().username(getUsername()).password(getPassword()).url(getUrl()).build();
    if (febpDBSrc != null && febpDBSrc instanceof HikariDataSource) {
      @SuppressWarnings("resource")
      HikariDataSource hikariDatsource = (HikariDataSource) febpDBSrc;
      hikariDatsource.setSchema(getSchema());
      hikariDatsource.setAutoCommit(false);
    }
    return febpDBSrc;
  }

  private String getSchema() {
    return febpProperties.getValue("spring.datasource.hikari.schema", "");
  }

  private String getUsername() {
    return febpProperties.getValue("spring.datasource.username", "");
  }

  private String getPassword() {
    return febpProperties.getValue("spring.datasource.password", "");
  }

  private String getUrl() {
    return febpProperties.getValue("spring.datasource.jdbc-url", "");
  }

  @Bean(name = "transactionManager")
  public JdbcTransactionManager batchTransactionManager(@Qualifier("febpDataSource") DataSource dataSource) {
    return new JdbcTransactionManager(dataSource);
  }

  @Bean(name = "febpBatchJobRepository")
  public JobRepository jobRepository(@Qualifier("febpDataSource") DataSource batchDataSource,
      @Qualifier("transactionManager") JdbcTransactionManager batchTransactionManager) throws Exception {
    JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
    factory.setDataSource(batchDataSource);
    factory.setTransactionManager(batchTransactionManager);
    factory.afterPropertiesSet();
    return factory.getObject();
  }

So shall I add hikariDatsource.setTransactionIsolation("ISOLATION_READ_COMMITTED"); inside batchDataSource method. Because this is where I am creating datasource for batch

[fmbenhassine]

no, you set it on the job repository factory bean: see setIsolationLevelForCreate.

[fmbenhassine]

I updated the separate answer with an example.

[fmbenhassine]

This is similar to Spring Batch - could not serialize access due to read/write dependencies among transactions.

You need to lower the isolation level of the transaction at the job repository to ISOLATION_READ_COMMITTED (or lower if needed). The reason is that the default value of ISOLATION_SERIALIZABLE is aggressive for setups where multiple jobs are launched simultaneously (or at a very high frequency).

Edit: based on previous comments

So shall I add hikariDatsource.setTransactionIsolation("ISOLATION_READ_COMMITTED"); inside batchDataSource method. Because this is where I am creating datasource for batch

no, you set it on the job repository factory bean: see setIsolationLevelForCreate

Here is an example:

  @Bean(name = "febpBatchJobRepository")
  public JobRepository jobRepository(@Qualifier("febpDataSource") DataSource batchDataSource,
      @Qualifier("transactionManager") JdbcTransactionManager batchTransactionManager) throws Exception {
    JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
    factory.setDataSource(batchDataSource);
    factory.setTransactionManager(batchTransactionManager);
    factory.setIsolationLevelForCreate("ISOLATION_REPEATABLE_READ");
    factory.afterPropertiesSet();
    return factory.getObject();
  }