spring-projectsGH-8577: Revise ImapIdleChannelAdapter logic

Fixes spring-projects#8577

When we process mail messages in async manner, it is possible that we end up
in a race condition situation where the next idle cycle closes the folder.

It is possible to reopen the folder, but feels better to block the current idle
cycle until we are done with the message and therefore keep folder opened.

* Deprecate `ImapIdleChannelAdapter.sendingTaskExecutor` in favor of an `ExecutorChannel`
as an output for this channel adapter or similar async hand-off downstream.
* Make use of `shouldReconnectAutomatically` as it is advertised for this channel adapter
* Optimize the proxy creation for message sending task

Author: artembilan

spring-integration-mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,13 +20,12 @@
 import java.time.Instant;
 import java.util.List;
 import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
 
 import jakarta.mail.Folder;
 import jakarta.mail.Message;
-import jakarta.mail.MessagingException;
 import org.aopalliance.aop.Advice;
 
 import org.springframework.aop.framework.ProxyFactory;
@@ -38,6 +37,7 @@
 import org.springframework.integration.transaction.IntegrationResourceHolder;
 import org.springframework.integration.transaction.IntegrationResourceHolderSynchronization;
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
+import org.springframework.messaging.MessagingException;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.TriggerContext;
@@ -78,12 +78,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
 
 	private boolean shouldReconnectAutomatically = true;
 
-	private Executor sendingTaskExecutor = Executors.newFixedThreadPool(1);
-
-	private boolean sendingTaskExecutorSet;
-
 	private List<Advice> adviceChain;
 
+	private Consumer<Object> messageSender;
+
 	private long reconnectDelay = DEFAULT_RECONNECT_DELAY; // milliseconds
 
 	private volatile ScheduledFuture<?> receivingTask;
@@ -104,20 +102,19 @@ public void setAdviceChain(List<Advice> adviceChain) {
 	}
 
 	/**
-	 * Specify an {@link Executor} used to send messages received by the
-	 * adapter.
+	 * Specify an {@link Executor} used to send messages received by the adapter.
 	 * @param sendingTaskExecutor the sendingTaskExecutor to set
+	 * @deprecated since 6.1 in favor of async hands-off downstream in the flow,
+	 * e.g. {@link  org.springframework.integration.channel.ExecutorChannel}.
 	 */
+	@Deprecated(since = "6.1", forRemoval = true)
 	public void setSendingTaskExecutor(Executor sendingTaskExecutor) {
-		Assert.notNull(sendingTaskExecutor, "'sendingTaskExecutor' must not be null");
-		this.sendingTaskExecutor = sendingTaskExecutor;
-		this.sendingTaskExecutorSet = true;
 	}
 
 	/**
 	 * Specify whether the IDLE task should reconnect automatically after
-	 * catching a {@link jakarta.mail.FolderClosedException} while waiting for messages. The
-	 * default value is <code>true</code>.
+	 * catching a {@link jakarta.mail.MessagingException} while waiting for messages. The
+	 * default value is true.
 	 * @param shouldReconnectAutomatically true to reconnect.
 	 */
 	public void setShouldReconnectAutomatically(boolean shouldReconnectAutomatically) {
@@ -148,6 +145,26 @@ public void setApplicationEventPublisher(ApplicationEventPublisher applicationEv
 		this.applicationEventPublisher = applicationEventPublisher;
 	}
 
+	@Override
+	@SuppressWarnings("unchecked")
+	protected void onInit() {
+		super.onInit();
+
+		Consumer<?> messageSenderToUse = new MessageSender();
+
+		if (!CollectionUtils.isEmpty(this.adviceChain)) {
+			ProxyFactory proxyFactory = new ProxyFactory(messageSenderToUse);
+			this.adviceChain.forEach(proxyFactory::addAdvice);
+			for (Advice advice : this.adviceChain) {
+				proxyFactory.addAdvice(advice);
+			}
+			messageSenderToUse = (Consumer<?>) proxyFactory.getProxy(this.classLoader);
+		}
+
+		this.messageSender = (Consumer<Object>) messageSenderToUse;
+	}
+
+
 	/*
 	 * Lifecycle implementation
 	 */
@@ -170,69 +187,53 @@ protected void doStop() {
 	public void destroy() {
 		super.destroy();
 		this.mailReceiver.destroy();
-		// If we're running with the default executor, shut it down.
-		if (!this.sendingTaskExecutorSet && this.sendingTaskExecutor != null) {
-			((ExecutorService) this.sendingTaskExecutor).shutdown();
+	}
+
+	private void publishException(Exception ex) {
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(new ImapIdleExceptionEvent(ex));
+		}
+		else {
+			logger.debug(() -> "No application event publisher for exception: " + ex.getMessage());
 		}
 	}
 
-	private Runnable createMessageSendingTask(Object mailMessage) {
-		Runnable sendingTask = prepareSendingTask(mailMessage);
+	private class MessageSender implements Consumer<Object> {
 
-		// wrap in the TX proxy if necessary
-		if (!CollectionUtils.isEmpty(this.adviceChain)) {
-			ProxyFactory proxyFactory = new ProxyFactory(sendingTask);
-			if (!CollectionUtils.isEmpty(this.adviceChain)) {
-				for (Advice advice : this.adviceChain) {
-					proxyFactory.addAdvice(advice);
-				}
-			}
-			sendingTask = (Runnable) proxyFactory.getProxy(this.classLoader);
+		MessageSender() {
 		}
-		return sendingTask;
-	}
 
-	private Runnable prepareSendingTask(Object mailMessage) {
-		return () -> {
-			@SuppressWarnings("unchecked")
-			org.springframework.messaging.Message<?> message =
+		@Override
+		public void accept(Object mailMessage) {
+			org.springframework.messaging.Message<?> messageToSend =
 					mailMessage instanceof Message
 							? getMessageBuilderFactory().withPayload(mailMessage).build()
-							: (org.springframework.messaging.Message<Object>) mailMessage;
+							: (org.springframework.messaging.Message<?>) mailMessage;
 
 			if (TransactionSynchronizationManager.isActualTransactionActive()
-					&& this.transactionSynchronizationFactory != null) {
+					&& ImapIdleChannelAdapter.this.transactionSynchronizationFactory != null) {
 
-				TransactionSynchronization synchronization = this.transactionSynchronizationFactory.create(this);
+				TransactionSynchronization synchronization =
+						ImapIdleChannelAdapter.this.transactionSynchronizationFactory.create(this);
 				if (synchronization != null) {
 					TransactionSynchronizationManager.registerSynchronization(synchronization);
 
-					if (synchronization instanceof IntegrationResourceHolderSynchronization
+					if (synchronization instanceof IntegrationResourceHolderSynchronization integrationSync
 							&& !TransactionSynchronizationManager.hasResource(this)) {
 
-						TransactionSynchronizationManager.bindResource(this,
-								((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder());
+						TransactionSynchronizationManager.bindResource(this, integrationSync.getResourceHolder());
 					}
 
 					Object resourceHolder = TransactionSynchronizationManager.getResource(this);
-					if (resourceHolder instanceof IntegrationResourceHolder) {
-						((IntegrationResourceHolder) resourceHolder).setMessage(message);
+					if (resourceHolder instanceof IntegrationResourceHolder integrationResourceHolder) {
+						integrationResourceHolder.setMessage(messageToSend);
 					}
 				}
 			}
-			sendMessage(message);
-		};
-	}
-
-	private void publishException(Exception ex) {
-		if (this.applicationEventPublisher != null) {
-			this.applicationEventPublisher.publishEvent(new ImapIdleExceptionEvent(ex));
+			sendMessage(messageToSend);
 		}
-		else {
-			logger.debug(() -> "No application event publisher for exception: " + ex.getMessage());
-		}
-	}
 
+	}
 
 	private class ReceivingTask implements Runnable {
 
@@ -246,10 +247,23 @@ public void run() {
 					ImapIdleChannelAdapter.this.idleTask.run();
 					logger.debug("Task completed successfully. Re-scheduling it again right away.");
 				}
-				catch (Exception ex) { //run again after a delay
-					logger.warn(ex, () -> "Failed to execute IDLE task. Will attempt to resubmit in "
-							+ ImapIdleChannelAdapter.this.reconnectDelay + " milliseconds.");
-					ImapIdleChannelAdapter.this.receivingTaskTrigger.delayNextExecution();
+				catch (Exception ex) {
+					if (ImapIdleChannelAdapter.this.shouldReconnectAutomatically
+							&& ex.getCause() instanceof jakarta.mail.MessagingException messagingException) {
+
+						//run again after a delay
+						logger.info(messagingException,
+								() -> "Failed to execute IDLE task. Will attempt to resubmit in "
+										+ ImapIdleChannelAdapter.this.reconnectDelay + " milliseconds.");
+						ImapIdleChannelAdapter.this.receivingTaskTrigger.delayNextExecution();
+					}
+					else {
+						logger.warn(ex,
+								"Failed to execute IDLE task. " +
+										"Won't resubmit since not a 'shouldReconnectAutomatically'" +
+										"or not a 'jakarta.mail.MessagingException'");
+						ImapIdleChannelAdapter.this.receivingTaskTrigger.stop();
+					}
 					publishException(ex);
 				}
 			}
@@ -274,21 +288,19 @@ public void run() {
 						Object[] mailMessages = ImapIdleChannelAdapter.this.mailReceiver.receive();
 						logger.debug(() -> "received " + mailMessages.length + " mail messages");
 						for (Object mailMessage : mailMessages) {
-							Runnable messageSendingTask = createMessageSendingTask(mailMessage);
 							if (isRunning()) {
-								ImapIdleChannelAdapter.this.sendingTaskExecutor.execute(messageSendingTask);
+								ImapIdleChannelAdapter.this.messageSender.accept(mailMessage);
 							}
 						}
 					}
 				}
-				catch (MessagingException ex) {
+				catch (jakarta.mail.MessagingException ex) {
 					logger.warn(ex, "error occurred in idle task");
 					if (ImapIdleChannelAdapter.this.shouldReconnectAutomatically) {
 						throw new IllegalStateException("Failure in 'idle' task. Will resubmit.", ex);
 					}
 					else {
-						throw new org.springframework.messaging.MessagingException(
-								"Failure in 'idle' task. Will NOT resubmit.", ex);
+						throw new MessagingException("Failure in 'idle' task. Will NOT resubmit.", ex);
 					}
 				}
 			}
@@ -298,16 +310,20 @@ public void run() {
 
 	private class ExceptionAwarePeriodicTrigger implements Trigger {
 
-		private volatile boolean delayNextExecution;
+		private final AtomicBoolean delayNextExecution = new AtomicBoolean();
+
+		private final AtomicBoolean stop = new AtomicBoolean();
 
 
 		ExceptionAwarePeriodicTrigger() {
 		}
 
 		@Override
 		public Instant nextExecution(TriggerContext triggerContext) {
-			if (this.delayNextExecution) {
-				this.delayNextExecution = false;
+			if (this.stop.getAndSet(false)) {
+				return null;
+			}
+			if (this.delayNextExecution.getAndSet(false)) {
 				return Instant.now().plusMillis(ImapIdleChannelAdapter.this.reconnectDelay);
 			}
 			else {
@@ -316,7 +332,11 @@ public Instant nextExecution(TriggerContext triggerContext) {
 		}
 
 		void delayNextExecution() {
-			this.delayNextExecution = true;
+			this.delayNextExecution.set(true);
+		}
+
+		void stop() {
+			this.stop.set(true);
 		}
 
 	}

spring-integration-mail/src/main/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -54,8 +54,6 @@ protected AbstractBeanDefinition doParse(Element element, ParserContext parserCo
 			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, txElement,
 					"synchronization-factory", "transactionSynchronizationFactory");
 		}
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor",
-				"sendingTaskExecutor");
 		AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
 		IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(null,
 				DomUtils.getChildElementByTagName(element, "transactional"), beanDefinition, parserContext);

spring-integration-mail/src/main/java/org/springframework/integration/mail/dsl/ImapIdleChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2022 the original author or authors.
+ * Copyright 2014-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -352,10 +352,11 @@ public ImapIdleChannelAdapterSpec transactional() {
 	 * Specify a task executor to be used to send messages to the downstream flow.
 	 * @param sendingTaskExecutor the sendingTaskExecutor.
 	 * @return the spec.
-	 * @see ImapIdleChannelAdapter#setSendingTaskExecutor(Executor)
+	 * @deprecated since 6.1 in favor of async hands-off downstream in the flow,
+	 * e.g. {@link  org.springframework.integration.channel.ExecutorChannel}.
 	 */
+	@Deprecated(since = "6.1", forRemoval = true)
 	public ImapIdleChannelAdapterSpec sendingTaskExecutor(Executor sendingTaskExecutor) {
-		this.target.setSendingTaskExecutor(sendingTaskExecutor);
 		return this;
 	}
 

spring-integration-mail/src/main/resources/org/springframework/integration/mail/config/spring-integration-mail.xsd

@@ -167,10 +167,11 @@
 					<xsd:attribute name="task-executor" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation><![CDATA[
-								Reference to a bean that implements
+								[DEPRECATED] Reference to a bean that implements
 								org.springframework.core.task.TaskExecutor which is used
 								to send Messages received by this adapter.
 								If not provided, the adapter uses a single-threaded executor.
+								Deprecated since 6.1 in favor of async hand-off downstream in the flow.
 							]]></xsd:documentation>
 							<xsd:appinfo>
 								<tool:annotation kind="ref">

spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -209,6 +209,7 @@ public void testIdleWithServerGuts(ImapMailReceiver receiver, boolean mapped, bo
 		adapter.setOutputChannel(channel);
 		adapter.setTaskScheduler(taskScheduler);
 		adapter.setReconnectDelay(1);
+		adapter.afterPropertiesSet();
 		adapter.start();
 		MimeMessage message =
 				GreenMailUtil.createTextEmail("Foo <foo@bar>", "Bar <bar@baz>", "Test Email", "foo\r\n",
@@ -698,6 +699,7 @@ public void testInitialIdleDelayWhenRecentIsSupported() throws Exception {
 		QueueChannel channel = new QueueChannel();
 		adapter.setOutputChannel(channel);
 		adapter.setReconnectDelay(1);
+		adapter.afterPropertiesSet();
 
 		ImapMailReceiver receiver = new ImapMailReceiver("imap:foo");
 		receiver.setCancelIdleInterval(1);

spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests-context.xml

@@ -78,8 +78,7 @@
 			store-uri="imap:foo"
 			channel="channel"
 			auto-startup="false"
-			should-delete-messages="true"
-			task-executor="executor">
+			should-delete-messages="true">
 		<mail:transactional synchronization-factory="syncFactory" />
 	</mail:imap-idle-channel-adapter>
 

spring-integration-mail/src/test/java/org/springframework/integration/mail/config/ImapIdleChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -194,7 +194,6 @@ public void transactionalAdapter() {
 		assertThat(receiverAccessor.getPropertyValue("shouldDeleteMessages")).isEqualTo(Boolean.TRUE);
 		assertThat(receiverAccessor.getPropertyValue("shouldMarkMessagesAsRead")).isEqualTo(Boolean.TRUE);
 		assertThat(adapterAccessor.getPropertyValue("errorChannel")).isNull();
-		assertThat(adapterAccessor.getPropertyValue("sendingTaskExecutor")).isEqualTo(context.getBean("executor"));
 		assertThat(adapterAccessor.getPropertyValue("adviceChain")).isNotNull();
 	}