daprog daprog - 3 months ago 44
Java Question

How properly define Spring Integration JpaOutboundGateway in Java Configuration?

I'm trying to replace Spring Integration XML configuration by JavaConfig and I can't properly define JpaOutboundGateway. In XML i have following bean definition:

<int-jpa:outbound-channel-adapter id="saveOffer" entity-manager-factory="entityManagerFactory" channel="channel5"
flush-size="100" entity-class="JobOffer">
<int-jpa:transactional transaction-manager="transactionManager"/>
</int-jpa:outbound-channel-adapter>


And it works fine. But when I'm trying to move it to java config bean I don't know how to define transaction management... I'm still getting errors connected with transactions.

@Bean
@ServiceActivator(inputChannel="channel5")
public JpaOutboundGatewayFactoryBean save() throws Exception {

JpaExecutor executor = new JpaExecutor(entityManager);
executor.setFlushSize(10);
executor.setEntityClass(JobOffer.class);

JpaOutboundGatewayFactoryBean factoryBean = new JpaOutboundGatewayFactoryBean();
factoryBean.setJpaExecutor(executor);
factoryBean.setProducesReply(false);

return factoryBean;
}


I was trying to use EntityManager and EntityManagerFactory in JpaExecutor. But in both cases exception is thrown:

2016-09-03 09:20:42.440 ERROR o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.integration.jpa.outbound.JpaOutboundGateway@1fc306d3]; nested exception is javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process 'merge' call
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:139)
at org.springframework.integration.endpoint.PollingConsumer.handleMessage(PollingConsumer.java:129)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:272)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.access$000(AbstractPollingEndpoint.java:58)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:190)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:186)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:353)
at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:55)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:51)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:344)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process 'merge' call
at org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:282)
at com.sun.proxy.$Proxy121.merge(Unknown Source)
at org.springframework.integration.jpa.core.DefaultJpaOperations.persistOrMerge(DefaultJpaOperations.java:248)
at org.springframework.integration.jpa.core.DefaultJpaOperations.merge(DefaultJpaOperations.java:215)
at org.springframework.integration.jpa.core.JpaExecutor.executeOutboundJpaOperation(JpaExecutor.java:253)
at org.springframework.integration.jpa.outbound.JpaOutboundGateway.handleRequestMessage(JpaOutboundGateway.java:81)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
... 19 more

Answer

This is one of the more complicated endpoints to wire up; we need to add an example for Java configuration to the reference manual; I opened a JIRA Issue.

You can always look at the XML parser to see what's needed.

In particular:

  1. the JpaExecutor should be defined as as @Bean.
  2. You need to add a transaction advice to the adapter's txAdviceChain.
  3. You can look at IntegrationNamespaceUtils.configureTransactionAttributes to see what's needed to wire up the transaction advice (transaction attribute, attribute source, TM, etc) wired into a TransactionInterceptor.
Comments