1

I'm writing a batch that reads from one table that returns let's say X. And for every X I'm making rest api requests in the processor and I'm writing the response to the another table in the writer. For every X I'm writing 0..N records to another table.

I want to read 10 items per page, and I want to process these items using 5 threads after processing I want to write response data to the db.

For this architecture, I'm using HibernatePagingItemReader for ItemReader with a pageSize of 10 for example. And I'm using AsyncItemProcessor for processing and AsyncItemWriter for write part. And my chunkSize is same with pageSize.

My problem is, I'm logging X's id in the ItemProcessor (also in the itemread listener) and in different threads I see the same X getting read/processed. How can I fix this?

My code looks something like this;


public class MyItemReader extends HibernatePagingItemReader<X> {

  public MyItemReader(SessionFactory sessionFactory, Integer pageSize) { // Reader constructor. 
    setSessionFactory(sessionFactory);
    setName("MyItemReader");
    setPageSize(pageSize);
    setQueryString(getQuery());
    setParameterValues(getQueryParams());
    setSaveState(false);
  }
}

@Bean
public ItemReader<X> myItemReader() {
    return new MyItemReader(sessionFactory, pageSize);
}


@SneakyThrows
@Bean
@StepScope
public AsyncItemProcessor<X, Y> myItemProcessor() {
    var asyncItemProcessor = new AsyncItemProcessor<X, Y>();
    asyncItemProcessor.setDelegate(new MyItemProcessor());
    asyncItemProcessor.setTaskExecutor(taskExecutor());
    asyncItemProcessor.afterPropertiesSet();
    return asyncItemProcessor;
}

@SneakyThrows
@Bean
public AsyncItemWriter<Y> myItemWriter() {
    var asyncItemWriter = new AsyncItemWriter<Y>();
    asyncItemWriter.setDelegate(new MyItemWriter(repository));
    asyncItemWriter.afterPropertiesSet();
    return asyncItemWriter;
}


@Bean
public TaskExecutor taskExecutor() {
    SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
    executor.setConcurrencyLimit(getNumberOfThreads());
    executor.setThreadNamePrefix("Thread-");
    return executor;
}

@Bean
public Step myStoreStep() {
    return baseStep("myStep")
        .<X, Future<Y>>chunk(chunkSize)
        .reader(myItemReader())
        .processor(myItemProcessor())
        .writer(myItemWriter())
        .taskExecutor(taskExecutor())
        .build();
}

Update: I changed the question title because the problem stems from HibernatePagingItemReader not the AsyncItemProcessor

10
  • The hibernate reader is documented to be thread-safe between calls to open/close. But have you tried to wrap it in a SynchronizedItemStreamReader? If not, please give it a try and share your feedback. Nov 24, 2022 at 10:32
  • I tried to wrap MyItemReader with SynchronizedItemStreamReader but it didn't change anything. I can still see same id logged more than once. java public ItemReader<X> myItemReader() { var sync = new SynchronizedItemStreamReader<X>(); sync.setDelegate(new MyItemReader(sessionFactory, pageSize)); sync.afterPropertiesSet(); return sync; }
    – Omeerfk
    Nov 24, 2022 at 10:44
  • For starters I would change your return types for the methods to the actual types and not an interface. Doing only some toplevel interface is problematic with callbacks. Also does your MyItemWriter and MyItemProcessor also need some callbacks? If so they need to be proper beans as well.
    – M. Deinum
    Nov 24, 2022 at 12:20
  • Thanks for your input but my problem is HibernatePagingItemReader reading same data more than once. I tried returning MyItemReader itself but that didnt change anything
    – Omeerfk
    Nov 24, 2022 at 13:37
  • Please provide a minimal example that reproduces the issue and I will try to help. Nov 24, 2022 at 14:11

1 Answer 1

0

As mentioned in the comment, I think the listener is misleading and makes you think items are processed more than once while they are not. Here is a sample output I get when running your sample:


  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::                (v2.6.3)

2022-12-06 09:10:40.973  INFO 49865 --- [           main] c.f.s.SpringBatchPagingReaderApplication : No active profile set, falling back to default profiles: default
2022-12-06 09:10:41.391  INFO 49865 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data JPA repositories in DEFAULT mode.
2022-12-06 09:10:41.534  INFO 49865 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 132 ms. Found 1 JPA repository interfaces.
2022-12-06 09:10:41.588  INFO 49865 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
2022-12-06 09:10:41.601  INFO 49865 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
2022-12-06 09:10:41.757  INFO 49865 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2022-12-06 09:10:41.767  INFO 49865 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2022-12-06 09:10:41.834  INFO 49865 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Starting...
2022-12-06 09:10:41.989  INFO 49865 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Start completed.
2022-12-06 09:10:42.129  INFO 49865 --- [           main] o.hibernate.jpa.internal.util.LogHelper  : HHH000204: Processing PersistenceUnitInfo [name: default]
2022-12-06 09:10:42.181  INFO 49865 --- [           main] org.hibernate.Version                    : HHH000412: Hibernate ORM core version 5.6.4.Final
2022-12-06 09:10:42.329  INFO 49865 --- [           main] o.hibernate.annotations.common.Version   : HCANN000001: Hibernate Commons Annotations {5.1.2.Final}
2022-12-06 09:10:42.456  INFO 49865 --- [           main] org.hibernate.dialect.Dialect            : HHH000400: Using dialect: org.hibernate.dialect.H2Dialect
2022-12-06 09:10:42.966  INFO 49865 --- [           main] o.h.e.t.j.p.i.JtaPlatformInitiator       : HHH000490: Using JtaPlatform implementation: [org.hibernate.engine.transaction.jta.platform.internal.NoJtaPlatform]
2022-12-06 09:10:42.973  INFO 49865 --- [           main] j.LocalContainerEntityManagerFactoryBean : Initialized JPA EntityManagerFactory for persistence unit 'default'
2022-12-06 09:10:43.264  WARN 49865 --- [           main] o.s.b.a.batch.JpaBatchConfigurer         : JPA does not support custom isolation levels, so locks may not be taken when launching Jobs
2022-12-06 09:10:43.266  INFO 49865 --- [           main] o.s.b.c.r.s.JobRepositoryFactoryBean     : No database type set, using meta data indicating: H2
2022-12-06 09:10:43.407  INFO 49865 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : No TaskExecutor has been set, defaulting to synchronous executor.
2022-12-06 09:10:43.589  INFO 49865 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
2022-12-06 09:10:43.589  INFO 49865 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application.errorChannel' has 1 subscriber(s).
2022-12-06 09:10:43.589  INFO 49865 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
2022-12-06 09:10:43.599  INFO 49865 --- [           main] c.f.s.SpringBatchPagingReaderApplication : Started SpringBatchPagingReaderApplication in 2.909 seconds (JVM running for 3.477)
2022-12-06 09:10:43.600  INFO 49865 --- [           main] o.s.b.a.b.JobLauncherApplicationRunner   : Running default command line with: []
2022-12-06 09:10:43.649  INFO 49865 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=myJob]] launched with the following parameters: [{}]
2022-12-06 09:10:43.684  INFO 49865 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [myStep]
2022-12-06 09:10:43.807  INFO 49865 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [myStep] executed in 123ms
2022-12-06 09:10:43.812  INFO 49865 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=myJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 139ms
2022-12-06 09:10:43.896  INFO 49865 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=myJob]] launched with the following parameters: [{time=1670314243893}]
2022-12-06 09:10:43.900  INFO 49865 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [myStep]
2022-12-06 09:10:43.907  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=cd600a96-18e4-48b2-ad3d-507781ff56e4, name=Entity-49-processed, date=2022-12-06T09:10:43.907659)
2022-12-06 09:10:43.907  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=d83d45e0-b2dd-4a39-a485-3ed2b9006d3d, name=Entity-47-processed, date=2022-12-06T09:10:43.907659)
2022-12-06 09:10:43.907  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=0a81a593-15aa-4b94-a2a7-289b26cb7cb6, name=Entity-48-processed, date=2022-12-06T09:10:43.907659)
2022-12-06 09:10:43.908  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=3a0c3a43-8cff-416b-ae56-dee6eef55510, name=Entity-44-processed, date=2022-12-06T09:10:43.908829)
2022-12-06 09:10:43.908  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=ef579965-754b-469f-938d-07d100ca604f, name=Entity-45-processed, date=2022-12-06T09:10:43.908829)
2022-12-06 09:10:43.908  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=520d30e1-0949-4d11-94f4-2e2639dcbf8a, name=Entity-46-processed, date=2022-12-06T09:10:43.908829)
2022-12-06 09:10:43.909  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=60a7ec08-5d2c-40e9-a74c-4c49ec564439, name=Entity-41-processed, date=2022-12-06T09:10:43.909266)
2022-12-06 09:10:43.909  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=301d0531-7789-4bc4-866c-a57edf9b0b4d, name=Entity-43-processed, date=2022-12-06T09:10:43.909266)
2022-12-06 09:10:43.909  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=1a141f30-53bf-4a3d-958c-832719efb9dd, name=Entity-42-processed, date=2022-12-06T09:10:43.909266)
2022-12-06 09:10:43.919  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=8fd7ef68-a5f6-49f2-a63b-4e00db9743a6, name=Entity-40-processed, date=2022-12-06T09:10:43.919006)
2022-12-06 09:10:43.919  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=ab1cf7b8-16f7-40a0-a796-d452100f95c7, name=Entity-39-processed, date=2022-12-06T09:10:43.919291)
2022-12-06 09:10:43.919  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=f01dbb6d-4de4-425d-ad92-636de972f46b, name=Entity-38-processed, date=2022-12-06T09:10:43.919423)
2022-12-06 09:10:43.919  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=5539c0ed-09a4-4bb7-bbfd-8e25c23d4100, name=Entity-37-processed, date=2022-12-06T09:10:43.919760)
2022-12-06 09:10:43.919  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=a04658ec-95df-45bf-97d6-0737e11c5e08, name=Entity-36-processed, date=2022-12-06T09:10:43.919872)
2022-12-06 09:10:43.919  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=e6482b80-7e6a-46b2-bb1d-ff3adca3ef6f, name=Entity-35-processed, date=2022-12-06T09:10:43.919963)
2022-12-06 09:10:43.923  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=a7b8d0ba-5bb8-4f09-9100-e4b0340fdb09, name=Entity-34-processed, date=2022-12-06T09:10:43.923216)
2022-12-06 09:10:43.923  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=853af3e1-ea1b-406a-b672-4910b9c2d1fc, name=Entity-33-processed, date=2022-12-06T09:10:43.923413)
2022-12-06 09:10:43.923  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=ec5e260e-ebd7-49e1-bbff-d7283ad660c6, name=Entity-32-processed, date=2022-12-06T09:10:43.923511)
2022-12-06 09:10:43.923  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=78789cdd-0dfa-417c-80b3-197424062cae, name=Entity-31-processed, date=2022-12-06T09:10:43.923901)
2022-12-06 09:10:43.924  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=5be18675-c5d9-435b-b896-1caac625da9e, name=Entity-30-processed, date=2022-12-06T09:10:43.924022)
2022-12-06 09:10:43.924  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=a46d8233-ac18-4dea-a173-c1c314d2cec8, name=Entity-29-processed, date=2022-12-06T09:10:43.924164)
2022-12-06 09:10:43.925  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=f918ed0b-5d66-4eca-b110-af0ed99419f7, name=Entity-28-processed, date=2022-12-06T09:10:43.925106)
2022-12-06 09:10:43.925  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=00ae2964-d5e7-43fa-acc2-eb5c1d471b57, name=Entity-27-processed, date=2022-12-06T09:10:43.925260)
2022-12-06 09:10:43.925  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=0d3f4907-d8e5-488b-a81f-ec48dfcd78a0, name=Entity-26-processed, date=2022-12-06T09:10:43.925350)
2022-12-06 09:10:43.928  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=e6277d2a-c42c-4bb9-b547-ac2bbec529cb, name=Entity-25-processed, date=2022-12-06T09:10:43.928304)
2022-12-06 09:10:43.928  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=6af2be3a-d37f-4c8e-9e84-0f0e8dd345a1, name=Entity-24-processed, date=2022-12-06T09:10:43.928496)
2022-12-06 09:10:43.928  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=2fe5047d-a372-4f7a-accb-7a92ab47badf, name=Entity-23-processed, date=2022-12-06T09:10:43.928595)
2022-12-06 09:10:43.929  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=e5bacc63-7e30-4f86-9f0f-58969e48627e, name=Entity-22-processed, date=2022-12-06T09:10:43.929452)
2022-12-06 09:10:43.929  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=1dbbcbe3-e7e8-4764-985b-5f4a460842f8, name=Entity-21-processed, date=2022-12-06T09:10:43.929629)
2022-12-06 09:10:43.929  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=a314e534-57f2-48bc-86fb-b8bbc864430c, name=Entity-20-processed, date=2022-12-06T09:10:43.929722)
2022-12-06 09:10:43.931  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=a632d843-cd28-451e-9ca7-9a67d267a7ed, name=Entity-19-processed, date=2022-12-06T09:10:43.931264)
2022-12-06 09:10:43.931  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=f8d0ce64-e806-46be-9412-f8907a68735e, name=Entity-18-processed, date=2022-12-06T09:10:43.931455)
2022-12-06 09:10:43.931  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=06041655-8260-4637-81f2-f35d6aac6aec, name=Entity-17-processed, date=2022-12-06T09:10:43.931546)
2022-12-06 09:10:43.932  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=d703c25d-cb63-42e5-8d5b-e699ea410cf7, name=Entity-16-processed, date=2022-12-06T09:10:43.932924)
2022-12-06 09:10:43.933  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=ad7d58de-4962-49ea-b9d2-37aa3ec1ccca, name=Entity-15-processed, date=2022-12-06T09:10:43.933123)
2022-12-06 09:10:43.933  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=c907025d-a863-4701-ad7e-5d7c298e774b, name=Entity-14-processed, date=2022-12-06T09:10:43.933246)
2022-12-06 09:10:43.934  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=eba0c9b7-3aac-4e68-aa29-cf63643e3fac, name=Entity-13-processed, date=2022-12-06T09:10:43.934154)
2022-12-06 09:10:43.934  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=fef8acdd-9aa8-4a04-8489-2d2726c5495a, name=Entity-12-processed, date=2022-12-06T09:10:43.934340)
2022-12-06 09:10:43.934  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=a8933475-68bd-48a2-a486-7967a8c2ea0c, name=Entity-11-processed, date=2022-12-06T09:10:43.934464)
2022-12-06 09:10:43.936  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=c307bdfa-8946-434b-9ed2-6e4a4a8fa88d, name=Entity-10-processed, date=2022-12-06T09:10:43.936282)
2022-12-06 09:10:43.936  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=2335a872-a7f3-4e28-884b-5d930b3c3429, name=Entity-9-processed, date=2022-12-06T09:10:43.936448)
2022-12-06 09:10:43.936  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=3efd8445-ed6f-403f-a26d-6786700196ad, name=Entity-8-processed, date=2022-12-06T09:10:43.936534)
2022-12-06 09:10:43.937  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=24326a3b-9005-458c-a273-4cbdff51f43c, name=Entity-7-processed, date=2022-12-06T09:10:43.937383)
2022-12-06 09:10:43.937  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=b32b4bdc-52b9-4a27-8d32-1108d0db8abb, name=Entity-6-processed, date=2022-12-06T09:10:43.937544)
2022-12-06 09:10:43.937  INFO 49865 --- [       Thread-2] c.f.springbatchpagingreader.MyProcessor  : Thread-2 processed: MyEntity(id=50e1d017-ab8b-4870-88d5-984e519037ae, name=Entity-5-processed, date=2022-12-06T09:10:43.937665)
2022-12-06 09:10:43.939  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=5a7c0341-212f-442d-af5a-ce1517bf5761, name=Entity-4-processed, date=2022-12-06T09:10:43.939247)
2022-12-06 09:10:43.939  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=b880c519-7d73-4851-9f19-48ce6da7bbe0, name=Entity-3-processed, date=2022-12-06T09:10:43.939438)
2022-12-06 09:10:43.939  INFO 49865 --- [       Thread-3] c.f.springbatchpagingreader.MyProcessor  : Thread-3 processed: MyEntity(id=7a639650-6568-4d81-a3a9-fa7401ca1c6a, name=Entity-2-processed, date=2022-12-06T09:10:43.939539)
2022-12-06 09:10:43.940  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=da662b27-0297-472c-9f5a-21be2141ec38, name=Entity-1-processed, date=2022-12-06T09:10:43.940009)
2022-12-06 09:10:43.940  INFO 49865 --- [       Thread-1] c.f.springbatchpagingreader.MyProcessor  : Thread-1 processed: MyEntity(id=6774f169-d6af-48d9-af1f-ec9897a7e6cb, name=Entity-0-processed, date=2022-12-06T09:10:43.940138)
2022-12-06 09:10:43.947  INFO 49865 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [myStep] executed in 47ms
2022-12-06 09:10:43.949  INFO 49865 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=myJob]] completed with the following parameters: [{time=1670314243893}] and the following status: [COMPLETED] in 52ms

I just changed your processor like follows:

@Slf4j
public class MyProcessor implements ItemProcessor<MyEntity, MyEntity> {
    @Override
    public MyEntity process(MyEntity myEntity) throws Exception {
        myEntity.setName(myEntity.getName() + "-processed");
        myEntity.setDate(LocalDateTime.now());
        log.info(Thread.currentThread().getName() + " processed: {}", myEntity);
        return myEntity;
    }
}

and removed the listener from the step definition.

According to the output, items are processed concurrently by different threads, and there are no items processed twice. Some items are processed out of order (which is expected), like Entity-41 and Entity-44. The total number of items is also correct (50 items in total). I have a similar output with a dozen executions.

So as mentioned in the comment, I cannot see how the same item could be read by two different threads because the AbstractPagingItemReader#doRead method is synchronized on a lock. I'm not sure if the behaviour you are expecting from the listener is actually what you think is really happening (ie only one thread at a time is calling afterRead, even though you are using a concurrent map in the listener) to draw the initial conclusion.

3
  • Again, thanks for your reply. I added listener to log the item info, I don't really need to use it. But removing it doesn't change my problem. I updated my repository with removed listener. And after removing I ran the app again (this is the log: gist.github.com/FarukKaradeniz/…). You can see some items are in the processor more that once(Entity-2) and some of them are never in the processor(Entity-8).
    – Omeerfk
    Dec 6, 2022 at 12:10
  • In the real batch application I'm working on (at work). Some items takes about couple of minutes to process thats why I added Thread.sleep() to this example to simulate.
    – Omeerfk
    Dec 6, 2022 at 12:12
  • I tried couple more times but I'm getting inconsistent results. In my local environment when I run the code I can see multiple reads but when I run the same code on docker container it works as I expected. I'm just hoping I won't get the same problem on production :)
    – Omeerfk
    Dec 6, 2022 at 14:04

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.