Safely read expunged IMAP messages

If application is connected to a Domino mail server via IMAP, it can happen from time to time that a message is expunged.
This leads to a `MessageRemovedException` when calling `IMAPMessage#getSubject`.
And although debug is set to false this again leads to a `MessageException` and the whole integration flow stops.

* Check for `isExpunged()` before logging a filtered message

* Add `@author`
* Fix debug message for expunged to reflect reality
* Clean up new unit tests a bit

**Cherry-pick to `5.4.x`**

Author: domi-87

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -64,6 +64,7 @@
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Dominik Simmen
  */
 public abstract class AbstractMailReceiver extends IntegrationObjectSupport implements MailReceiver, DisposableBean {
 
@@ -551,11 +552,15 @@ private MimeMessage[] filterMessagesThruSelector(Message[] messages) throws Mess
 						this.selectorExpression.getValue(this.evaluationContext, message, Boolean.class))) {
 					filteredMessages.add(message);
 				}
-				else {
-					String subject = message.getSubject();
-					this.logger.debug(() ->
-							"Fetched email with subject '" + subject
-									+ "' will be discarded by the matching filter and will not be flagged as SEEN.");
+				else if (this.logger.isDebugEnabled()) {
+					if (message.isExpunged()) {
+						this.logger.debug("Expunged message discarded and will not be further processed.");
+					}
+					else {
+						String subject = message.getSubject();
+						this.logger.debug("Fetched email with subject '" + subject +
+								"' will be discarded by the matching filter and will not be flagged as SEEN.");
+					}
 				}
 			}
 			else {

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

@@ -28,6 +28,7 @@
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.lang.reflect.Field;
@@ -91,6 +92,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Alexander Pinske
+ * @author Dominik Simmen
  */
 @SpringJUnitConfig
 @ContextConfiguration(
@@ -309,6 +311,53 @@ public void receiveAndMarkAsReadDontDeleteFiltered() throws Exception {
 		verify(receiver, times(0)).deleteMessages(Mockito.any());
 	}
 
+	@Test
+	public void receiveAndDebugIsDisabledNotLogFiltered() throws Exception {
+		AbstractMailReceiver receiver = new ImapMailReceiver();
+
+		LogAccessor logger = spy(TestUtils.getPropertyValue(receiver, "logger", LogAccessor.class));
+		new DirectFieldAccessor(receiver).setPropertyValue("logger", logger);
+		when(logger.isDebugEnabled()).thenReturn(false);
+
+		Message msg1 = mock(MimeMessage.class);
+		Message msg2 = mock(MimeMessage.class);
+		Expression selectorExpression = new SpelExpressionParser().parseExpression("false");
+		receiver.setSelectorExpression(selectorExpression);
+		receiveAndMarkAsReadDontDeleteGuts(receiver, msg1, msg2);
+		verify(logger, times(2)).isDebugEnabled();
+		verify(msg1, never()).isExpunged();
+		verify(msg2, never()).isExpunged();
+		verify(msg1, never()).getSubject();
+		verify(msg2, never()).getSubject();
+		verify(logger, never()).debug(Mockito.startsWith("Expunged message received"));
+		verify(logger, never()).debug(org.mockito.ArgumentMatchers.contains("will be discarded by the matching filter"));
+	}
+
+	@Test
+	public void receiveExpungedAndNotExpungedLogFiltered() throws Exception {
+		AbstractMailReceiver receiver = new ImapMailReceiver();
+
+		LogAccessor logger = spy(TestUtils.getPropertyValue(receiver, "logger", LogAccessor.class));
+		new DirectFieldAccessor(receiver).setPropertyValue("logger", logger);
+		when(logger.isDebugEnabled()).thenReturn(true);
+
+		Message msg1 = mock(MimeMessage.class);
+		Message msg2 = mock(MimeMessage.class);
+		given(msg1.isExpunged()).willReturn(true);
+		given(msg1.getSubject()).willReturn("msg1");
+		given(msg2.getSubject()).willReturn("msg2");
+		Expression selectorExpression = new SpelExpressionParser().parseExpression("false");
+		receiver.setSelectorExpression(selectorExpression);
+		receiveAndMarkAsReadDontDeleteGuts(receiver, msg1, msg2);
+		verify(logger, times(2)).isDebugEnabled();
+		verify(msg1).isExpunged();
+		verify(msg2).isExpunged();
+		verify(msg1, never()).getSubject();
+		verify(msg2).getSubject();
+		verify(logger).debug(Mockito.startsWith("Expunged message discarded"));
+		verify(logger).debug(org.mockito.ArgumentMatchers.contains("'msg2' will be discarded by the matching filter"));
+	}
+
 
 	@Test
 	public void receiveMarkAsReadAndDelete() throws Exception {