title: Spring Batch JDBCPagingItemReader not partitioning equally for each thread
body: <p>this is my first question here. I am working on a spring batch and I am using step partitioning for processing 70K records. For testing I am using 1021 records and found that the partitioning not happening equally for each thread. I am using JDBCPagingItemReader with 5 threads. The distribution should be </p>
  
  <p>Thread 1 - 205</p>
  
  <p>Thread 2 - 205</p>
  
  <p>Thread 3 - 205</p>
  
  <p>Thread 4 - 205</p>
  
  <p>Thread 5 - 201</p>
  
  <p>But unfortunately this is not happening and I am getting the below record distribution among threads</p>
  
  <p>Thread 1 - 100</p>
  
  <p>Thread 2 - 111</p>
  
  <p>Thread 3 - 100</p>
  
  <p>Thread 4 - 205</p>
  
  <p>Thread 5 - 200</p>
  
  <p>Total 716 records and 305 records are skipped while partitioning. I really don't have any clue what is happening. Could you please look at the below configurations and let me know am I missing anything? Thanks in advance for your help.</p>
  
  <pre><code>&lt;import resource="../config/batch-context.xml" /&gt;
  &lt;import resource="../config/database.xml" /&gt;
  
  &lt;job id="partitionJob"  xmlns="http://www.springframework.org/schema/batch"&gt;
  
      &lt;step id="masterStep" parent="abstractPartitionerStagedStep"&gt;
  
          &lt;partition step="slave" partitioner="rangePartitioner"&gt;
              &lt;handler grid-size="5" task-executor="taskExecutor"/&gt;
          &lt;/partition&gt;
  
      &lt;/step&gt;
  
  &lt;/job&gt;
  &lt;bean id="abstractPartitionerStagedStep" abstract="true"&gt;
      &lt;property name="listeners"&gt;
          &lt;list&gt;
              &lt;ref bean="updatelistener" /&gt;
          &lt;/list&gt;
      &lt;/property&gt;
  
  &lt;/bean&gt;
  &lt;bean id="updatelistener" 
        class="com.test.springbatch.model.UpdateFileCopyStatus" &gt;
  &lt;/bean&gt;
  &lt;!-- Jobs to run --&gt;
  &lt;step id="slave" xmlns="http://www.springframework.org/schema/batch"&gt;
      &lt;tasklet&gt;
          &lt;chunk reader="pagingItemReader" writer="flatFileItemWriter"
              processor="itemProcessor" commit-interval="1" retry-limit="0" skip-limit="100"&gt;
          &lt;skippable-exception-classes&gt;
              &lt;include class="java.lang.Exception"/&gt;
          &lt;/skippable-exception-classes&gt;
          &lt;/chunk&gt;    
      &lt;/tasklet&gt;
  &lt;/step&gt;
  
  &lt;bean id="rangePartitioner" class="com.test.springbatch.partition.RangePartitioner"&gt; 
      &lt;property name="dataSource" ref="dataSource" /&gt;
  &lt;/bean&gt;
  
  &lt;bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" &gt;
  &lt;property name="corePoolSize" value="5"/&gt;
  &lt;property name="maxPoolSize" value="5"/&gt;
  &lt;property name="queueCapacity" value="100" /&gt;
  &lt;property name="allowCoreThreadTimeOut" value="true"/&gt;
  &lt;property name="keepAliveSeconds" value="60" /&gt; 
  &lt;/bean&gt;
  
  &lt;bean id="itemProcessor" class="com.test.springbatch.processor.CaseProcessor" scope="step"&gt;
      &lt;property name="threadName" value="#{stepExecutionContext[name]}" /&gt;
  &lt;/bean&gt;
  
  &lt;bean id="pagingItemReader"
      class="org.springframework.batch.item.database.JdbcPagingItemReader"
      scope="step"&gt;
      &lt;property name="dataSource" ref="dataSource" /&gt;
      &lt;property name="queryProvider"&gt;
          &lt;bean
              class="org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean"&gt;
              &lt;property name="dataSource" ref="dataSource" /&gt;
              &lt;property name="selectClause" value="SELECT *" /&gt;
              &lt;property name="fromClause" value="FROM ( SELECT CASE_NUM ,CASE_STTS_CD, UPDT_TS,SBMT_OFC_CD,
                          SBMT_OFC_NUM,DSTR_CHNL_CD,APRV_OFC_CD,APRV_OFC_NUM,SBMT_TYP_CD, ROW_NUMBER() 
                          OVER(ORDER BY CASE_NUM) AS rownumber FROM TSMCASE WHERE PROC_IND ='N' ) AS data" /&gt;
              &lt;property name="whereClause" value="WHERE rownumber BETWEEN :fromRow AND :toRow " /&gt;
              &lt;property name="sortKey" value="CASE_NUM" /&gt;
          &lt;/bean&gt;
      &lt;/property&gt;
      &lt;!--  Inject via the ExecutionContext in rangePartitioner  --&gt;
      &lt;property name="parameterValues"&gt;
          &lt;map&gt;
              &lt;entry key="fromRow" value="#{stepExecutionContext[fromRow]}" /&gt;
              &lt;entry key="toRow" value="#{stepExecutionContext[toRow]}" /&gt;
          &lt;/map&gt;
      &lt;/property&gt;
      &lt;property name="pageSize" value="100" /&gt; 
      &lt;property name="rowMapper"&gt;
          &lt;bean class="com.test.springbatch.model.CaseRowMapper" /&gt;
      &lt;/property&gt;
  &lt;/bean&gt;
  
  &lt;bean id="flatFileItemWriter" class="com.test.springbatch.writer.FNWriter" scope="step" &gt;
  &lt;/bean&gt;
  </code></pre>
  
  <p>Here the partitioner code</p>
  
  <pre><code>public class OffRangePartitioner implements Partitioner {
  
  private String officeLst;
  private double splitvalue;
  private DataSource dataSource;
  private static Logger LOGGER = Log4JFactory.getLogger(OffRangePartitioner.class);
  private static final int INDENT_LEVEL = 6;
  
  public String getOfficeLst() {
      return officeLst;
  }
  
  public void setOfficeLst(final String officeLst) {
      this.officeLst = officeLst;
  }
  
  public void setDataSource(DataSource dataSource) {
      this.dataSource = dataSource;
  }
  
  public OfficeRangePartitioner() {
      super();
      final GlobalProperties globalProperties = GlobalProperties.getInstance();
      splitvalue = Double.parseDouble(globalProperties.getProperty("springbatch.part.splitvalue"));
  }
  
  @Override
  public Map&lt;String, ExecutionContext&gt; partition(int threadSize) {
      FormattedTraceHelper.formattedTrace(LOGGER,"Partition method in OffRangePartitioner class Start",INDENT_LEVEL, Level.INFO_INT);
      final Session currentSession = HibernateUtil.getSessionFactory(HibernateConstants.DB2_DATABASE_NAME).getCurrentSession();
  
      Query queryObj;
      double count = 0.0;
  
      final Transaction transaction = currentSession.beginTransaction();
      queryObj = currentSession.createQuery(BatchConstants.PARTITION_CNT_QRY);
  
      if (queryObj.iterate().hasNext()) {
          count = Double.parseDouble(queryObj.iterate().next().toString());
      }
  
      int fromRow = 0;
      int toRow = 0;
      ExecutionContext context;
  
      FormattedTraceHelper.formattedTrace(LOGGER,"Count of total records submitted for processing &gt;&gt; " + count, INDENT_LEVEL, Level.DEBUG_INT);
      int gridSize = (int) Math.ceil(count / splitvalue);
      FormattedTraceHelper.formattedTrace(LOGGER,"Total Grid size based on the count &gt;&gt; " + gridSize, INDENT_LEVEL, Level.DEBUG_INT);
      Map&lt;String, ExecutionContext&gt; result = new HashMap&lt;String, ExecutionContext&gt;();
  
      for (int threadCount = 1; threadCount &lt;= gridSize; threadCount++) {
          fromRow = toRow + 1;
          if (threadCount == gridSize || gridSize == 1) {
              toRow = (int) count;
          } else {
              toRow += splitvalue;
          }
          context = new ExecutionContext();
          context.putInt("fromRow", fromRow);
          context.putInt("toRow", toRow);
          context.putString("name", "Processing Thread" + threadCount);
          result.put("partition" + threadCount, context);
          FormattedTraceHelper.formattedTrace(LOGGER, "Partition number &gt;&gt; "
                  + threadCount + " from Row#: " + fromRow + " to Row#: "
                  + toRow, INDENT_LEVEL, Level.DEBUG_INT);
  
      }
      if (transaction != null) {
          transaction.commit();
      }
      FormattedTraceHelper.formattedTrace(LOGGER,
              "Partition method in OffRangePartitioner class End",
              INDENT_LEVEL, Level.INFO_INT);
      return result;
  }
  </code></pre>
  
  <p>}</p>
  
  <p>Today, I have tested the same batch with 1056 records with Spring Framework log debug on.</p>
  
  <p>PAGE SIZE 100</p>
  
  <pre><code>SELECT * FROM (
          SELECT CASE_NUM, CASE_STTS_CD, UPDT_TS,SBMT_OFC_CD, SBMT_OFC_NUM, DSTR_CHNL_CD, 
              APRV_OFC_CD, APRV_OFC_NUM,SBMT_TYP_CD, ROW_NUMBER() OVER(ORDER BY CASE_NUM) AS rownumber 
          FROM TCASE 
          WHERE **SECARCH_PROC_IND ='P'**
      ) AS data 
  WHERE 
      rownumber BETWEEN :fromRow AND :toRow 
  ORDER BY 
      rownumber ASC 
  FETCH FIRST 100 ROWS ONLY
  </code></pre>
  
  <p>We are updating the SECARCH_PROC_IND ='P' flag to 'C' once each record processed. We are using ROWNUM in the main query to partition the records based on SECARCH_PROC_IND ='P' and the ROWNUM getting changed once the SECARCH_PROC_IND ='P' flag updated to 'C'by any threads.</p>
  
  <p>Looks like this is the issue.</p>