Spring integration recipient-list-router issue

See also questions close to this topic

• To read large file and store them in pojo and inturn in arraylist

  Hi I have my program structure - in a spring boot project as below:

  in a Service class: File 1 - loaded in arraylist1(list of pojos)

  File 2 -- loaded in arraylist2(list of pojos)

  File 3- loaded in arraylist3(list of pojos)

  input file --- parsed and loaded in arraylist. output Arraylist for (iterate input file- arraylist){

  //for output field 1
  for(iterate in file1){
      if field available in file1 - assign output column
  else
  reject record..
  }
  
  //for output field 2
  for(iterate in file2){
      if field available in file2 - assign output column
  else
  reject record..
  }
  
  
  //for output field 3
  for(iterate in file3){
      if field available in file3 - assign output column
  else
  reject record..
  }
  
  assign to other output fields from input fields..
  
  output field 4=inputfield 4
  output field 5=inputfield 5
  output field 6=inputfield 6
  output field 7=inputfield 7
  output field 8=inputfield 8
  

  outputList.add(output pojo) }

  So while reading the File 2 which is of 2 gb , the process hungs or throws Out of memory error. Completely stuck with this,Please help with this problem. Thank you
• How to handle exception and skip the wrong csv line as well?

  I am reading a csv file using Spring batch. I am reading the content of csv and writing this content to the database in accordance with one of the entity class.

  Now there could be certain lines in csv that are wrong and don't match the POJO attributes. To handle this I configured my Step as follows:

  Step step = stepBuilderFactory.get("CSV-Step")
                  .<Book, Book>chunk(100)
                  .faultTolerant()
                  .skip(FlatFileParseException.class)
                  .skipLimit(1)
                  .reader(itemReader)
                  .writer(itemWriter)
                  .build();
  

  It basically skips the line which causes FlatFileParseException and goes on with subsequent lines. Now I also want to log the lines for which parsing could not be done. For this in my GlobalExceptionHandler annotated with @ControllerAdvice I made following method:

  @OnReadError
  public void handleCsvParseException(FlatFileParseException ex, Throwable throwable) {
      logger.error("! FlatFileParseException, line  is: " + ex.getLineNumber());
      logger.error("! FlatFileParseException, input is: " + ex.getInput());
      logger.error("! Message: " + throwable.getMessage());
      logger.error("! Cause: " + throwable.getCause());
  }
  

  The thing is that this method is not being called because i have the skip configuration in my Step. How can i ignore the unwanted lines i.e. skipping unwanted line but at the same time log information about them. I would appreciate any kind of help.
• Why are interface projections much slower than constructor projections and entity projections in Spring Data JPA with Hibernate?

  I've been wondering which kind of projections should I use, so I did a little test, which covered 5 types of projections (based on docs: https://docs.spring.io/spring-data/jpa/docs/current/reference/html/#projections):

  1. Entity projection

  This is just a standard findAll() provided by Spring Data repository. Nothing fancy here.

  Service:

  List<SampleEntity> projections = sampleRepository.findAll();
  

  Entity:

  @Entity
  @Table(name = "SAMPLE_ENTITIES")
  public class SampleEntity {
      @Id
      private Long id;
      private String name;
      private String city;
      private Integer age;
  }
  

  2. Constructor projection

  Service:

  List<NameOnlyDTO> projections = sampleRepository.findAllNameOnlyConstructorProjection();
  

  Repository:

  @Query("select new path.to.dto.NameOnlyDTO(e.name) from SampleEntity e")
  List<NameOnlyDTO> findAllNameOnlyConstructorProjection();
  

  Data transfer object:

  @NoArgsConstructor
  @AllArgsConstructor
  public class NameOnlyDTO {
      private String name;
  }
  

  3. Interface projection

  Service:

  List<NameOnly> projections = sampleRepository.findAllNameOnlyBy();
  

  Repository:

  List<NameOnly> findAllNameOnlyBy();
  

  Interface:

  public interface NameOnly {
      String getName();
  }
  

  4. Tuple projection

  Service:

  List<Tuple> projections = sampleRepository.findAllNameOnlyTupleProjection();
  

  Repository:

  @Query("select e.name as name from SampleEntity e")
  List<Tuple> findAllNameOnlyTupleProjection();
  

  5. Dynamic projection

  Service:

  List<DynamicProjectionDTO> projections = sampleRepository.findAllBy(DynamicProjectionDTO.class);
  

  Repository:

  <T> List<T> findAllBy(Class<T> type);
  

  Data transfer object:

  public class DynamicProjectionDTO {
  
      private String name;
  
      public DynamicProjectionDTO(String name) {
          this.name = name;
      }
  }
  

  
  Some additional info:

  The project was built using gradle spring boot plugin (version 2.0.4), which uses Spring 5.0.8 under the hood. Database: H2 in memory.

  Results:

  Entity projections took 161.61 ms on average out of 100 iterations.
  Constructor projections took 24.84 ms on average out of 100 iterations.
  Interface projections took 252.26 ms on average out of 100 iterations.
  Tuple projections took 21.41 ms on average out of 100 iterations.
  Dynamic projections took 23.62 ms on average out of 100 iterations.
  -----------------------------------------------------------------------
  One iteration retrieved (from DB) and projected 100 000 objects.
  -----------------------------------------------------------------------
  

  Notes:

  It is understandable that retrieving entities takes some time. Hibernate tracks these objects for changes, lazy loading and so on.

  Constructor projections are really fast and have no limitations on the DTO side, but require manual object creation in @Query annotation.

  Interface projections turned out to be really slow. See question.

  Tuple projections were the fastest, but are not the most convinient to play with. They need an alias in JPQL and the data has to be retrieved by calling .get("name") instead of .getName().

  Dynamic projections look pretty cool and fast, but must have exactly one constructor. No more, no less. Otherwise Spring Data throws an exception, because it doesn't know which one to use (it takes constructor parameters to determine which data to retrieve from DB).

  Question:

  Why interface projections take longer than retrieving entities? Each interface projection returned is actually a proxy. Is it so expensive to create that proxy? If so, doesn't it defeat the main purpose of projections (since they are meant to be faster than entities)? Other projections look awesome tho. I would really love some insight on this. Thank you.
• When I use spring integrationi, I can't forwad a request through firewall

  Our company want build a soft gateway to controll request. and our company alreay have a hardware firewall. I try to use spring integration to forward a http request. the code is very simple,only have xml file.

  <int:channel id="searchRequestChannel" />
  <int:channel id="searchReplyChannel" />
  <int-http:inbound-gateway id="searchRequestInboundGateway"
                        supported-methods="GET"
                        request-channel="searchRequestChannel"
                        reply-channel="searchReplyChannel"
                        path="/services/outrequest"
                        reply-timeout="5000"
  />
  <int-http:outbound-gateway id="searchServiceGateway"
                         http-method="GET"
                         request-channel="searchRequestChannel"
                         url="${forward.path}"
                         extract-request-payload="false"
                         expected-response-type="java.lang.String"
                         reply-timeout="5000"
                         charset="UTF-8"
   />
  

  the code runs well in the other network ,but when it runs in our company's inner network,some website shows time out error. But if I vist the website directly, it runs well. So I want to known if the spring integration add something to my reqeust?
• Generate Spring-integration daily statistics report

  Have a spring integration application where files are routed from a folder to S3 buckets using s3-outbound-channel-adapter. If the file processed successfully, then file will be moved to corresponding target-bucket. if any error , file move to error bucket via error channel.

  Have to generate a daily statistics report in a text file containing below details.

  Total no of files processed: Total success : Total Error:

  Would like to know how to get no of files processed successfully/error. Is there any way to achieve this requirement.

  Any suggestion or example would be helpful.

  Gone through the DefaultMessageChannelMetrics and Micrometer Integration in documentation. Not sure it will help my requirement.

  Have separate gateway and adapter to process success and error files.

  Success :

  <int-aws:s3-outbound-gateway id="s3FileMover"
          request-channel="filesOutS3GateWay"
          reply-channel="filesOutS3ChainChannel"
          transfer-manager="transferManager"
          bucket-expression = "headers.TARGET_PATH"
          key-expression="headers.file_name"
          command="UPLOAD">
          <int-aws:request-handler-advice-chain>
              <ref bean="retryAdvice" />
          </int-aws:request-handler-advice-chain>
      </int-aws:s3-outbound-gateway>
  

  Error :

  <int-aws:s3-outbound-channel-adapter id="filesErrorS3Mover"
              channel="filesErrorS3MoverChannel"
              transfer-manager="transferManager"
              bucket="${aws.s3.error.bucket}"
               key-expression="headers.TARGET + '/' + headers.file_name"
              upload-metadata-provider = "fileMetaDataProvider"
              command="UPLOAD">
              <int-aws:request-handler-advice-chain>
                  <bean class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice">
                      <property name="onSuccessExpressionString" value="payload.delete()"/>
                  </bean>
              </int-aws:request-handler-advice-chain>
  
• Controller mapping are not logging at startup

  I have a requirement to upgrade the existing code which runs on jdk 8 to jdk 11 After upgrading it the controller mappings are not showing up during startup.In this application we have defined some inbound gateways for integration and we have also defined few rest controllers, They where getting logging when it was on jdk 1.8 but they are not getting logged after i upgraded.Is there any way those logs are printed. we are using spring boot 2.1.0 Release, jdk 11 spring integration. The first image is the the code running on the jdk 8 and the second image is the one running on jdk 11. Here is pomfile which we are using http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0

  <groupId>com.oms.integration</groupId>
  <artifactId>oms-integration</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>
  
  <name>oms-integration</name>
  <description>Integration between OMS and other systems</description>
  
  <parent>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-parent</artifactId>
      <version>2.1.0.RELEASE</version>
      <relativePath /> <!-- lookup parent from repository -->
  </parent>
  
  <properties>
      <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      <java.version>11</java.version>
  </properties>
  
  <dependencies>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-actuator</artifactId>
      </dependency>
  
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-integration</artifactId>
      </dependency>
  
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-devtools</artifactId>
      </dependency>
  
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-web</artifactId>
      </dependency>
  
      <dependency>
          <groupId>org.springframework.integration</groupId>
          <artifactId>spring-integration-http</artifactId>
      </dependency>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-test</artifactId>
          <scope>test</scope>
      </dependency>
      <dependency>
          <groupId>org.apache.commons</groupId>
          <artifactId>commons-lang3</artifactId>
      </dependency>
      <dependency>
          <groupId>org.springframework.integration</groupId>
          <artifactId>spring-integration-xml</artifactId>
      </dependency>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-configuration-processor</artifactId>
          <optional>true</optional>
      </dependency>
      <dependency>
          <groupId>xmlunit</groupId>
          <artifactId>xmlunit</artifactId>
          <version>1.5</version>
      </dependency>
  </dependencies>
  
  <build>
      <plugins>
          <plugin>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-compiler-plugin</artifactId>
              <configuration>
                  <release>${java.version}</release>
              </configuration>
              <dependencies>
                  <dependency>
                      <groupId>org.ow2.asm</groupId>
                      <artifactId>asm</artifactId>
                      <version>6.2</version>
                  </dependency>
              </dependencies>
          </plugin>
          <plugin>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-maven-plugin</artifactId>
          </plugin>
      </plugins>
  </build>
  

  
• SimpleMessageListenerContainer failed to reconnect on ems broker failover

  I have 2 jms listener applications one application uses DefaultMessageListenerContainerand other application uses SimpleMessageListenerContainer

  The application with DefaultMessageListenerContainer was able to reconnect on EMS broker failover, here is the log

  Setup of JMS message listener invoker failed for destination 'SOME.TOPIC.T' - trying to recover. Cause: Connection is closed
  Could not refresh JMS Connection for destination 'SOME.TOPIC.T' - retrying using FixedBackOff{interval=5000, currentAttempts=0, maxAttempts=unlimited}. Cause: Failed to connect to any server at: tcp://localhost:8285,tcp://localhost:8286
  Could not refresh JMS Connection for destination 'SOME.TOPIC.T' - retrying using FixedBackOff{interval=5000, currentAttempts=1, maxAttempts=unlimited}. Cause: Failed to connect to any server at: tcp://localhost:8285,tcp://localhost:8286
  

  The application with SimpleMessageListenerContainer was not able to reconnect on EMS broker failover, here is the log

  [TIBCO EMS TCPLink Reader (Server-109365)] INFO  org.springframework.jms.listener.SimpleMessageListenerContainer - Trying to recover from JMS Connection exception: javax.jms.JMSException: Connection has been terminated
  [TIBCO EMS TCPLink Reader (Server-109366)] ERROR org.springframework.jms.listener.SimpleMessageListenerContainer - Encountered non-recoverable JMSException
  javax.jms.JMSException: Connection has been terminated
      at com.tibco.tibjms.Tibjmsx.buildException(Tibjmsx.java:659) ~[tibjms-8.3.0.jar!/:?]
      at com.tibco.tibjms.TibjmsConnection._invokeOnExceptionCallback(TibjmsConnection.java:2118) [tibjms-8.3.0.jar!/:?]
      at com.tibco.tibjms.TibjmsConnection._onDisconnected(TibjmsConnection.java:2487) [tibjms-8.3.0.jar!/:?]
      at com.tibco.tibjms.TibjmsConnection$ServerLinkEventHandler.onEventDisconnected(TibjmsConnection.java:367) [tibjms-8.3.0.jar!/:?]
      at com.tibco.tibjms.TibjmsxLinkTcp$LinkReader.work(TibjmsxLinkTcp.java:328) [tibjms-8.3.0.jar!/:?]
      at com.tibco.tibjms.TibjmsxLinkTcp$LinkReader.run(TibjmsxLinkTcp.java:259) [tibjms-8.3.0.jar!/:?]
  

  Both are connected to same broker instance, how to make the SimpleMessageListenerContainer listener to reconnect on JMS Exceptions?
• How to create a ready to use library which will have all connection details to JMs?

  I need to create a library which will have all the connection to JMS. Ex.. connection factory, mesaage convertor, JMSListenerContaineRFactory, JMSTemplete This library i will give to clients so that they can use it to send mesaage to JMS queue

  I am wondering how could this be done.i am confused, will this library have a main class? How will the beans be initialized?
• Monitor embedded ActiveMQ broker

  I have a spring web application deployed on Tomcat. My application uses activemq-spring-5.14.5 jar which gives it the capability to define queues/destination and the broker is getting created by ActiveMQ at vm://localhost(embedded).

  Now I want to monitor my jms broker, queue and the producers and consumers though either activemq web console or Hawtio. I have searched the forums but I couldn't figure out what to be done.

  Most of them talks about standalone or separate ActiveMQ installation.

  Any help will be useful.
• How to read messages from topic with multiple consumer?

  I read from one topic with 10 consumer and send these messages to one queue. When I send 50 messages to topic by the jmeter, queue has 500 messages. So, each consumer read same messages from topic and after send to queue. Can each consumer read different messages from topic?

  Thank you so much.

  JmsConfig.java

  @Configuration
  @EnableJms
  @ComponentScan(basePackages = "com.jms.config")
  public class JmsConfig {
      String BROKER_URL = "tcp://localhost:61616";
      String BROKER_USERNAME = "admin";
      String BROKER_PASSWORD = "admin";
  
      @Bean
      public ActiveMQConnectionFactory connectionFactory(){
          ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
          connectionFactory.setBrokerURL(BROKER_URL);
          connectionFactory.setPassword(BROKER_USERNAME);
          connectionFactory.setUserName(BROKER_PASSWORD);
          return connectionFactory;
      }
  
      @Bean
      public JmsTemplate jmsTemplate(){
          JmsTemplate template = new JmsTemplate();
          template.setConnectionFactory(connectionFactory());
          template.setPubSubDomain(true);
          return template;
      }
  
      @Bean
      public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
          DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
          factory.setConnectionFactory(connectionFactory());
          factory.setConcurrency("1-10");
          factory.setPubSubDomain(true);
          return factory;
      }
  

  JmsSender.java

  @Service
  public class JmsSender{
  
      private JmsTemplate jmsTemplate;
  
      @Value("#{appProperties.toQueueName}")
      private String queueName;
  
      @Autowired
      private ApplicationContextUtil applicationContextUtil;
  
      public void send(String rawData){
          getJmsTemplate().send(queueName, new MessageCreator() {
              @Override
              public Message createMessage(Session session) throws JMSException {
                   return session.createObjectMessage(rawData);
              }
          });
      }
      public JmsTemplate getJmsTemplate(){
          if (jmsTemplate == null){
              jmsTemplate = (JmsTemplate) applicationContextUtil.getBeanFromAppContext("jmsForQueue");
          }
          return  jmsTemplate;
      }
  }
  

  Worker.java

  @Component
  public class Worker {
  
      @Autowired
      private JmsSender jmsSender;
  
      @JmsListener(destination = "#{appProperties.fromTopicName}")
      public String receiveMessageFromTopic(final String jsonMessage) throws JMSException {
          System.out.println("Received message " + jsonMessage);
          jmsSender.send(jsonMessage);
          return response;
      }
  }
  

  QueueConfig.xml

  <?xml version="1.0" encoding="UTF-8"?>
     <beans xmlns="http://www.springframework.org/schema/beans"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xmlns:aop="http://www.springframework.org/schema/aop"
     xmlns:context="http://www.springframework.org/schema/context"
     xmlns:tx="http://www.springframework.org/schema/tx"
     xmlns:mvc="http://www.springframework.org/schema/mvc"
     xmlns:task="http://www.springframework.org/schema/task"
     xmlns:amq="http://activemq.apache.org/schema/core"
     xmlns:cache="http://www.springframework.org/schema/cache"
     xmlns:int="http://www.springframework.org/schema/integration"
     xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
     xmlns:jms="http://www.springframework.org/schema/jms"
     xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
     http://www.springframework.org/schema/aop
     http://www.springframework.org/schema/aop/spring-aop-4.3.xsd
     http://www.springframework.org/schema/context
     http://www.springframework.org/schema/context/spring-context-4.3.xsd
     http://www.springframework.org/schema/tx
     http://www.springframework.org/schema/tx/spring-tx-4.3.xsd
     http://activemq.apache.org/schema/core
     http://activemq.apache.org/schema/core/activemq-core-5.4.0.xsd
     http://www.springframework.org/schema/task
     http://www.springframework.org/schema/task/spring-task-4.3.xsd
     http://www.springframework.org/schema/mvc
     http://www.springframework.org/schema/mvc/spring-mvc-4.3.xsd
     http://www.springframework.org/schema/cache
     http://www.springframework.org/schema/cache/spring-cache.xsd
     http://www.springframework.org/schema/integration
     http://www.springframework.org/schema/integration/spring-integration.xsd
     http://www.springframework.org/schema/integration/jms
     http://www.springframework.org/schema/integration/jms/spring-integration-jms-4.3.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd"
     default-lazy-init="false">     
  
     <bean id="brokerUrl" class="java.lang.String">
         <constructor-arg value="#{appProperties.queueUrl}"/>
     </bean>
  
     <amq:connectionFactory id="amqConnectionFactory" brokerURL="#brokerUrl" dispatchAsync="true"/>
  
     <bean id="connectionFactory1" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
         <constructor-arg ref="amqConnectionFactory"/>
         <property name="maxConnections" value="#{appProperties.maxConnections}"/>
         <property name="idleTimeout" value="#{appProperties.idleTimeout}"/>
         <property name="maximumActiveSessionPerConnection" value = "10"/> 
  
     </bean>
  
     <bean id="jmsForQueue" class="org.springframework.jms.core.JmsTemplate">
         <constructor-arg ref="connectionFactory1"/>
     </bean>
  
     <bean id="jSONQueue" class="org.apache.activemq.command.ActiveMQQueue">
         <constructor-arg value="#{appProperties.toQueueName}"/>
     </bean>
  </beans>
  
• How to consume two subscriber (subscribed to same query) to one?

  I have two subscribers, subscribed to same query, but I want only one to consume it. How can I achieve it using map or something else?
• In pub/sub model, does publisher create number of copies equal to number of consumers?

  In pub/sub model, does publisher create number of copies of messages/events equal to number of consumers and publishes? OR is it possible to configure publisher in such a away that it generates only one message ?
• How to use jms connection factory to send multiple messages

  I am trying to create a library to send messages to JMS queue. The obejctive of creating this library is that apps who want to use it need worry about creating connection, and storing details about queue.

  I have two files

  Class MyMQConfiguration {
  
       private void init(){
         /*In this function i create 
   connectionFactory, JMS templete 
    inside a function*/
       }
  
   /*this function return jms tempalte created above*/
  
    public JmsTemplate getJMSTemplate(){
         return jmsTemplate;
        }
  
  }
  
  
  
  
  
  Class SendMessage{
  
  JmsTemplate jmsTemplate;
  
     SendMesaage(){ 
         /* this  constructor creates object of.          MyMQConfiguration and gets the template */
      }
  
  Public void sendMessage(String msg){
    jmsTemplate.convertAndSend(msg);
      }
  
  }
  

  This is a maven project and an executable jar will be created of it. Say MyJMSProject-1.0.jar

  Now, i want to use this jar in a different project to send message to queue

  Say i have below two classes in that project

  class Producer1{
  
  public void postMesaage(){
     SendMessage sendMsg = new.        SendMesssage();
      String msg ="XYZ";
      sendMsg.send(msg);
  }
  
  }
  
  Class Producer2{
  public void postMesaage(){
         SendMessage sendMsg = new.    SendMesssage();
        String msg ="XYZ";
        sendMsg.send(msg);
      }
  }
  

  My concern is, each producer class that I have created, creates SendMessage object which initilizes and creates new connection factory n all everytime I.e How can I reuse connection instead of creating new everytime. I.e how can i use the same Jmstemplate for sending message in different classes?
• How to set up ibmmq connection using jmstemplate in a springboot web application

  Can anyone suggest how to setup initial context and create new ibm mq connections, sessions,senders and consumers when using jmstemplate in a springboot web application. I have done a lot of research on this but did not find any full example of sending and receiving messages asynchronously using JMSTemplate in springboot application.
• Only one message get dequeued when configuring JmsTemplate with SESSION_TRANSACTED

  I am trying to configure JmsTemplate in transactional manner, so when error occures durring the processing or writing, i want the messages to be visible in the queue again. I have the below configurations which seems to works fine when error occures i.e. if error is thrown durring the writing or processing, the messages are "sent" back to the queue. But the problem is when everything is OK (i.e. no errors are thrown), then only one message is consumed, processed and dequeued. So if i have 10 messages in the queue, only one is dequeued and the job is marked as COMPLETED. If i configure JmsTemplate with

  jmsTemplate.setSessionTransacted(false);
  

  I have all messages consumed, processed and written, but of course I am loosing the transactional manner. The question is how to get all mesasges transactionaly consumed and processed and written?

  I have the following JmsTemplate configuration

  @Bean
  public JmsTemplate jmsTemplate() {
      JmsTemplate jmsTemplate = new JmsTemplate(activeMQConnectionFactory());
      jmsTemplate.setDefaultDestinationName(applicationProperties.getQueue());
      jmsTemplate.setDefaultDestination(queue());
      jmsTemplate.setReceiveTimeout(applicationProperties.getReceiveTimeout());
      jmsTemplate.setSessionTransacted(true);
      jmsTemplate.setSessionAcknowledgeMode(Session.SESSION_TRANSACTED);
      return jmsTemplate;
  }
  

  which is passed to my JmsItemReader

  @Bean
  public ItemReader<MyPojoClass> myJmsItemReader() throws Exception {
  
      JmsItemReader<MyPojoClass> myJmsItemReader= new JmsItemReader<>();
      myJmsItemReader.setItemType(MyPojoClass.class);
      myJmsItemReader.setJmsTemplate(jmsTemplate);
      myJmsItemReader.afterPropertiesSet();
      return myJmsItemReader;
  
  }
  

  and below is my tasklet configuration

  @Bean(name = "receiveProcessAndWrite")
  public TaskletStep receiveProcessAndWrite() throws Exception {
  
      return this.stepBuilderFactory.get("receiveProcessAndWrite")
              .transactionManager(transactionManager)
              .<MyPojoClass, MyPojoClass>chunk(1000)
              .readerIsTransactionalQueue()
              .reader(myJmsItemReader())
              .processor(myprocessor())
              .writer(updateStatusInDBandWrite())
              .listener(new ChunkCountListener())
              .build();
  
  }