Spring Batch Job Restart: What should be the correct values for chunk and readerPageSize with respect to grid size.

[PSHREYASHOLLA]

This is posted here as I did not get a response on https://stackoverflow.com/questions/78199004/restarting-a-failed-job-is-not-processing-the-failed-chunk-data-again-continua

This is my step definition for spring batch partition approach,

@Bean public Step step1() { 
   return new StepBuilder("step1", jobRepository)
                            .partitioner(slaveStep().getName(), partitioner())
                            .step(slaveStep())
                            .gridSize(febpEmployeeTaxCalculationBatchProperties.getTaxCalculationStep1PartitionGridSize())
                            .taskExecutor(actStmntTaskExecutor())
                           .build();
  }

    // slave step
    @Bean
    public Step slaveStep() 
    {
        
        try {
            return new StepBuilder("slaveStep", jobRepository)
                    .<EmployeeDetail, EmployeeTaxDetail>chunk(febpEmployeeTaxCalculationBatchProperties.getTaxCalculationStep1ReaderPageSize(),transactionManager)
                    .reader(pagingItemReader(null,null))
                    .processor(processor())
                    .writer(customerItemWriter())
                .taskExecutor(actStmntTaskExecutor())
                    .build();
        } catch (Exception ex) {
            throw new RuntimeException("Error creating slave step: " + ex.getMessage());
        }
    }
 /**
   * Act stmnt task executor.
   *
   * @return the simple async task executor
   */
  public SimpleAsyncTaskExecutor actStmntTaskExecutor() {
    SimpleAsyncTaskExecutor acctStmtTaskExecuter = new SimpleAsyncTaskExecutor();
    acctStmtTaskExecuter.setConcurrencyLimit(febpEmployeeTaxCalculationBatchProperties.getTaxCalculationStep1TaskExecuterThreadConcurrencyLimit());
    acctStmtTaskExecuter.setThreadPriority(febpEmployeeTaxCalculationBatchProperties.getTaxCalculationStep1TaskExecuterThreadPriority());
    acctStmtTaskExecuter.setThreadNamePrefix("FEBP_TAX_CALCULATION_GEN");
    return acctStmtTaskExecuter;
  }

So here for restart to work,

1)Recommendation for restart is not using TaskExecuter. So should we not use TaskExecuter as mentioned above in step1() and slaveStep()? How to achieve concurrency ?

2. So only SimpleAsyncTaskExecutor needs to be used? If yes is it at step or slavestep?

3)How should the chunk size, gridSize and readers pagesize and fetchsiz be calculated? As I see if chunk size is set to low values, restart is not re-writting failed data.

[PSHREYASHOLLA]

Team any update for this?

[fmbenhassine]

Gird size is a hint to the partitioner about the number of partitions you want to create.

The chunk size, as you might be familiar with, is the number of items that will be processed as a unit in a single transaction.

The page size is the number of items that the reader should read in a single page.

Those are different parameters and the is no relation between them.

It is up to you to find the best values for each. For example, if your dataset is composed of 10.000 records and the grid size = 10, then you will have 10 partitions with 1000 each. Now since each worker will be assigned 1000 records, you could set the page size = chunk size = 100 for instance (to have a single page read per chunk).

I am turning this into a discussion, and let's continue there.

EDIT: You might find the following SO threads useful

• https://stackoverflow.com/questions/58041942/utility-of-fetchsize-and-pagesize-in-reader-spring-batch
• https://stackoverflow.com/questions/46452527/difference-between-chunk-size-and-page-size-spring-batch

[PSHREYASHOLLA]

1. Now since each worker will be assigned 1000 records, you could set the page size = chunk size = 100 for instance (to have a single page read per chunk). So is pageSize related to chunk size?
2. /**
   • Consolidated statement step 1.
   • @return the step
     */
     @bean
     public Step consolidatedStatementStep1() {
     return new StepBuilder("FERA_CONSOLIDATED_STATEMENT_API_CALL_STEP", jobRepository)
     .partitioner(consolidatedStatementslaveStep().getName(), consolidatedStmntpartitioner())
     .step(consolidatedStatementslaveStep())
     .gridSize(feraConsolidatedStatementBatchProperties.getConsolidatedStatementBatchStep1PartitionGridSize())
     .taskExecutor(consolidatedStatementTaskExecutor()).build();
     }

/**

• Slave step.
• @return the step
  */
  @bean
  public Step consolidatedStatementslaveStep() {
  try {
  return new StepBuilder("consolidatedstatementslaveStep", jobRepository)
  .<ConsolidatedStatement, ConsolidatedStatement>chunk(
  feraConsolidatedStatementBatchProperties.getConsolidatedStatementStep1ReaderPageSize(),
  transactionManager)
  .reader(consolidatedStatementPagingItemReader(null, null))
  .processor(consolidatedStmntProcessor()).writer(consolidatedStatementItemWriter()).build();
  } catch (Exception ex) {
  throw new RuntimeException("Error creating consolidated statement slave step: " + ex.getMessage());
  }
  }

/**

• Consolidated statement task executor.
• @return the simple async task executor
  */
  public SimpleAsyncTaskExecutor consolidatedStatementTaskExecutor() {
  SimpleAsyncTaskExecutor acctStmtTaskExecuter = new SimpleAsyncTaskExecutor();
  acctStmtTaskExecuter.setThreadPriority(
  feraConsolidatedStatementBatchProperties.getConsolidatedStatementStep1TaskExecuterThreadPriority());
  acctStmtTaskExecuter.setThreadNamePrefix("FERA_CONSOLIDATED_STMNT_GEN");
  return acctStmtTaskExecuter;
  }
  So is it ok to keep .taskExecutor(consolidatedStatementTaskExecutor()) only in step and not slave step?

[fmbenhassine]

1. Now since each worker will be assigned 1000 records, you could set the page size = chunk size = 100 for instance (to have a single page read per chunk). So is pageSize related to chunk size?

No, there is no relation between those parameters, as mentioned in my previous comment.

So is it ok to keep .taskExecutor(consolidatedStatementTaskExecutor()) only in step and not slave step?

Yes, the worker step is not required to be multi-threaded, it can be a single threaded step. Therefore, there is no need to set a task executor on it.

[PSHREYASHOLLA]

Thanks let me the check the threads you shared. What about the 1st and 2nd question,
1)Recommendation for restart is not using TaskExecuter. So should we not use TaskExecuter as mentioned above in step1() and slaveStep()? How to achieve concurrency ?

2)So only SimpleAsyncTaskExecutor needs to be used? If yes is it at step or slavestep?

[fmbenhassine]

1)Recommendation for restart is not using TaskExecuter. So should we not use TaskExecuter as mentioned above in step1() and slaveStep()? How to achieve concurrency ?

Restartability is not compatible with concurrency. A multi-threaded step is not restartable. However, a partitioned step is restartable. So you can achieve restartability with parallelism, not concurrency.

2)So only SimpleAsyncTaskExecutor needs to be used? If yes is it at step or slavestep?

You are setting a task executor on both step1 and slaveStep. This means the worker step will be concurrent, and therefore, not restartable. You need to remove the task executor on the worker, and keep it on the manager.