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.
SynchronizedItemStreamReader
? If not, please give it a try and share your feedback.java public ItemReader<X> myItemReader() { var sync = new SynchronizedItemStreamReader<X>(); sync.setDelegate(new MyItemReader(sessionFactory, pageSize)); sync.afterPropertiesSet(); return sync; }
MyItemWriter
andMyItemProcessor
also need some callbacks? If so they need to be proper beans as well.