title: Spring Batch - JpaPagingItemReader - Works in MySQL - Duplicates in PostgreSQL
body: <p>Before Spring Batch job runs. I have a import table which contains all items that needs importing into our system. It is at this point verified to contain only items that does not exist in our system.</p>
  
  <p>Next I have a Spring Batch Job, which reads from this import table using JpaPagingItemReader.
  After work is done, it writes to db using ItemWriter.</p>
  
  <p>I run with page-size and chunk-size at 10000. 
  Now this works absolutely fine when running on MySQL innoDB. I can even use multiple threading and everything works fine.</p>
  
  <p>But now we are migrating to PostgreSQL, and the same Batch Job runs into some very strange problem 
  What happens is that it tries to insert duplicates into our system. This will naturally be rejected by unique index constraints and an error is thrown. 
  Since the import db table is verified to contain only non-existing before batch job starts, the only reason for this i can think of is that the JpaPagingItemReader reads some rows multiple times from import db table when i run on Postgres. But why would it do that?</p>
  
  <p>I have experimented with a lot of settings. Turning chunk and page-size down to around 100 only makes import slower, but still same error. Running single-thread instead of multiple threads only makes the error happen slightly later. 
  So what on earth could be the reason for my JpaPagingItemReader reading the same items multiple times only on PostgresSQL?
  The select statement backing the reader is simple, its a NamedQuery:</p>
  
  <pre><code>@NamedQuery(name = "ImportDTO.findAllForInsert",
              query = "select h from ImportDTO h where h.toBeImported = true")
  </code></pre>
  
  <p>Please also note that the toBeImported flag will not be altered by the batch job at all during runtime, so the results from this query should always return the same before, under and after the batch job.</p>
  
  <p>Any insights, tips or help is greatly appriciated!</p>
  
  <p>Here is Batch Config code:</p>
  
  <pre><code>import javax.persistence.EntityManagerFactory;
  
  import org.springframework.batch.core.Job;
  import org.springframework.batch.core.Step;
  import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
  import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  import org.springframework.batch.core.launch.support.RunIdIncrementer;
  import org.springframework.batch.core.launch.support.SimpleJobLauncher;
  import org.springframework.batch.core.repository.JobRepository;
  import org.springframework.batch.item.database.JpaPagingItemReader;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;
  import org.springframework.core.task.SimpleAsyncTaskExecutor;
  import org.springframework.core.task.TaskExecutor;
  
  @Configuration
  @EnableBatchProcessing
  public class BatchConfiguration {
  
      @Autowired
      private JobBuilderFactory jobBuilderFactory;
      @Autowired
      private StepBuilderFactory stepBuilderFactory;
      @Autowired
      private OrganizationItemWriter organizationItemWriter;
      @Autowired
      private EntityManagerFactory entityManagerFactory;
      @Autowired
      private OrganizationUpdateProcessor organizationUpdateProcessor;
      @Autowired
      private OrganizationInsertProcessor organizationInsertProcessor;
  
      private Integer organizationBatchSize = 10000;
      private Integer organizationThreadSize = 3;
      private Integer maxThreadSize = organizationThreadSize;
  
      @Bean
      public SimpleJobLauncher jobLauncher(JobRepository jobRepository) {
          SimpleJobLauncher launcher = new SimpleJobLauncher();
          launcher.setJobRepository(jobRepository);
          return launcher;
      }
  
      @Bean
      public JpaPagingItemReader&lt;ImportDTO&gt; findNewImportsToImport() throws Exception {
          JpaPagingItemReader&lt;ImportDTO&gt; databaseReader = new JpaPagingItemReader&lt;&gt;();
          databaseReader.setEntityManagerFactory(entityManagerFactory);
          JpaQueryProviderImpl&lt;ImportDTO&gt; jpaQueryProvider = new JpaQueryProviderImpl&lt;&gt;();
          jpaQueryProvider.setQuery("ImportDTO.findAllForInsert");
          databaseReader.setQueryProvider(jpaQueryProvider);
          databaseReader.setPageSize(organizationBatchSize);
          // must be set to false if multi threaded
          databaseReader.setSaveState(false);
          databaseReader.afterPropertiesSet();
          return databaseReader;
      }
  
      @Bean
      public JpaPagingItemReader&lt;ImportDTO&gt; findImportsToUpdate() throws Exception {
          JpaPagingItemReader&lt;ImportDTO&gt; databaseReader = new JpaPagingItemReader&lt;&gt;();
          databaseReader.setEntityManagerFactory(entityManagerFactory);
          JpaQueryProviderImpl&lt;ImportDTO&gt; jpaQueryProvider = new JpaQueryProviderImpl&lt;&gt;();
          jpaQueryProvider.setQuery("ImportDTO.findAllForUpdate");
          databaseReader.setQueryProvider(jpaQueryProvider);
          databaseReader.setPageSize(organizationBatchSize);
          // must be set to false if multi threaded
          databaseReader.setSaveState(false);
          databaseReader.afterPropertiesSet();
          return databaseReader;
      }
  
      @Bean
      public OrganizationItemWriter writer() throws Exception {
          return organizationItemWriter;
      }
  
      @Bean
      public StepExecutionNotificationListener stepExecutionListener() {
          return new StepExecutionNotificationListener();
      }
  
      @Bean
      public ChunkExecutionListener chunkListener() {
          return new ChunkExecutionListener();
      }
  
      @Bean
      public TaskExecutor taskExecutor() {
          SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
          taskExecutor.setConcurrencyLimit(maxThreadSize);
          return taskExecutor;
      }
  
      @Bean
      public Job importOrganizationsJob(JobCompletionNotificationListener listener) throws Exception {
          return jobBuilderFactory.get("importAndUpdateOrganizationJob")
                  .incrementer(new RunIdIncrementer())
                  .listener(listener)
                  .start(importNewOrganizationsFromImports())
                  .next(updateOrganizationsFromImports())
                  .build();
      }
  
      @Bean
      public Step importNewOrganizationsFromImports() throws Exception {
          return stepBuilderFactory.get("importNewOrganizationsFromImports")
                  .&lt;ImportDTO, Organization&gt; chunk(organizationBatchSize)
                  .reader(findNewImportsToImport())
                  .processor(organizationInsertProcessor)
                  .writer(writer())
                  .taskExecutor(taskExecutor())
                  .listener(stepExecutionListener())
                  .listener(chunkListener())
                  .throttleLimit(organizationThreadSize)
                  .build();
      }
  
  
      @Bean
      public Step updateOrganizationsFromImports() throws Exception {
          return stepBuilderFactory.get("updateOrganizationsFromImports")
                  .&lt;ImportDTO, Organization&gt; chunk(organizationBatchSize)
                  .reader(findImportsToUpdate())
                  .processor(organizationUpdateProcessor)
                  .writer(writer())
                  .taskExecutor(taskExecutor())
                  .listener(stepExecutionListener())
                  .listener(chunkListener())
                  .throttleLimit(organizationThreadSize)
                  .build();
      }
  }
  </code></pre>