Spring Integration JPA Inbound Adapter Sends Duplicate Mails

We have an application that stores mails/notifications in a database and sends a "reminder" email to the user, that they have received a new mail/notification.

We use Spring integration and JPA inbound adapter to poll the JPA entity for mails in the "PENDING" state and then send the mails. However, due to the fact that we have multiple instances running, occasionally two emails are sent to the user leading to confusion.

We tried to solve this by locking the entity with a PESSIMISTIC_WRITE transaction lock but it still occasionally happens.

I'm quite new to Spring integration so maybe I'm missing something obvious but here's the configuration:

@Configuration
@EnableConfigurationProperties(SendMailProperties::class, MailProperties::class)
class SendMailConfiguration(
    private val entityManagerFactory: EntityManagerFactory,
    private val transactionManager: TransactionManager
) {

    @Bean
    fun sendMailFlow(
        mailProperties: MailProperties,
        sendMailProperties: SendMailProperties,
        mailSenderMessageHandler: MessageHandler,
        mimeMailTransformer: MimeMailTransformer
    ): IntegrationFlow =
        queryPendingMails()
            .transform(::updateAndLockEntity)
            .transform(mimeMailTransformer::convertToMimeMessage)
            .enrichHeaders(Mail.headers().to(sendMailProperties.recipient).from(mailProperties.username))
            .log()
            .handle(mailSenderMessageHandler)
            .get()

    private fun queryPendingMails() = IntegrationFlow.from(
        Jpa.inboundAdapter(entityManagerFactory)
            .entityClass(JpaSecuremail::class.java)
            .maxResults(1)
            .jpaQuery(
                """
                select s from JpaSecuremail s 
                where s.mail.status = '$PENDING' and s.mail.direction = '$OUTBOUND'
                """.trimIndent()
            )
            .expectSingleResult(true),
        mailPoller()
    )

    private fun mailPoller(): (SourcePollingChannelAdapterSpec) -> Unit =
        { spec: SourcePollingChannelAdapterSpec ->
            spec.poller { factory: PollerFactory ->
                factory
                    .fixedDelay(SECONDS_10_MILLIS)
                    .transactional(
                        TransactionInterceptorBuilder(true)
                            .transactionManager(transactionManager)
                            .build()
                    )
            }
        }

    private fun updateAndLockEntity(secureMail: JpaSecuremail) =
        with(getTransactionalEntityManager(entityManagerFactory) as EntityManager) {
            lock(secureMail, PESSIMISTIC_WRITE)

            secureMail.mail.apply {
                status = SENT
                messageId = UUID.randomUUID().toString()
                sentDate = now()
            }
            persist(secureMail)
            flush()
            secureMail
        }

    @Bean
    fun mailSenderMessageHandler(mailSender: MailSender) =
        CircuitBreakerMessageHandler(Mail.outboundAdapter(mailSender))

    companion object {
        private const val SECONDS_10_MILLIS = 1000L
    }
}

Does anyone have an idea, why this happens? Is the transaction guaranteed over the integration process? Is the lock done correctly? Or maybe some pointers to an example (didn't find a matching one)? Any pointers help, thanks

Try to use nativeQuery() instead where you can specify FOR UPDATE. Your PESSIMISTIC_WRITE assumption is correct, but you do that a bit late already when Jpa.inboundAdapter() has done its job.

We probably can improve Spring Integration JPA module to accept that PESSIMISTIC_WRITE as an option for use-cases like this.