davsclaus commented on code in PR #28200:
URL: https://github.com/apache/camel/pull/28200#discussion_r4333294987
##########
components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java:
##########
@@ -298,6 +319,72 @@ private void releaseNotRouted(Queue<Exchange> notRouted) {
}
}
+ private void openFolder() throws MessagingException {
+ folder.open(Folder.READ_WRITE);
+ folderOpens.incrementAndGet();
+ }
+
+ /**
+ * The mail to mark once its exchange completed. When the folder was
opened again since the mail was retrieved (its
+ * connection dropped while the mail was routed), the mail object of the
poll is stale: for IMAP, the mail is looked
+ * up again by its UID. Returns null when it cannot be found (it was
expunged meanwhile, or the UIDVALIDITY of the
+ * folder changed): then no mail is marked, as marking the stale mail
object would mark another mail.
+ */
+ private Message mailToCommit(Message mail, Long uid, int batchOpen,
Exchange exchange) {
+ if (getEndpoint().getConfiguration().getProtocol().startsWith("pop3"))
{
+ // processCommit finds a POP3 mail again by its key
+ return mail;
+ }
+ try {
+ if (!folder.isOpen()) {
+ openFolder();
+ }
+ if (folderOpens.get() == batchOpen) {
+ return mail;
+ }
+ Message found = null;
+ if (uid != null && folder instanceof UIDFolder uidFolder &&
uidFolder.getUIDValidity() == batchUidValidity) {
+ found = uidFolder.getMessageByUID(uid);
+ }
+ if (found == null) {
+ LOG.warn("The mail folder {} was opened again after the mail
was retrieved, and the mail is no longer"
Review Comment:
Not blocking, two small things on this path:
- When `null` is returned, `processCommit` is skipped completely. So the
idempotent key of the mail is neither removed
(`idempotentRepositoryRemoveOnCommit=true` is the default) nor confirmed. For a
mail that was expunged, the key then stays in the repository for good; a shared
JDBC or Infinispan repository keeps the row. Settling the key here the same way
`processCommit` does would keep the repository clean.
- The WARN says the mail is "no longer found" also when `uid` is null (the
UID FETCH in `rememberUids` failed) or when the UIDVALIDITY changed. Saying
which of the three happened would help whoever reads the log.
##########
components/camel-mail/src/test/java/org/apache/camel/component/mail/MailConsumerFolderReopenedTest.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mail;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import jakarta.mail.Flags;
+import jakarta.mail.Folder;
+import jakarta.mail.Message;
+import jakarta.mail.Store;
+import jakarta.mail.internet.MimeMessage;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mail.Mailbox.MailboxUser;
+import org.apache.camel.component.mail.Mailbox.Protocol;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * When the connection of the IMAP folder drops while a mail of a poll is
routed, the consumer opens the folder again to
+ * mark the mail. The mail objects of the poll belong to the earlier open of
the folder: their message numbers are those
+ * of the mailbox at that time. When mails were expunged meanwhile (by another
client, or by the server), the message
+ * number designates another mail in the new open, which was marked instead: a
mail that was not routed was marked as
+ * seen (or deleted) and never consumed.
+ */
+class MailConsumerFolderReopenedTest extends CamelTestSupport {
+ private static final MailboxUser jane =
Mailbox.getOrCreateUser("MailConsumerFolderReopenedTest-jane", "secret");
+
+ private final DroppingProxy proxy = new
DroppingProxy(Mailbox.getPort(Protocol.imap));
+ private final AtomicBoolean dropOnSecond = new AtomicBoolean(true);
+
+ @Override
+ public void setupResources() throws Exception {
+ Mailbox.clearAll(jane);
+ proxy.start();
+ }
+
+ @AfterEach
+ void stopProxy() throws IOException {
+ proxy.close();
+ }
+
+ @Test
+ void mailNotRoutedIsNotMarkedWhenTheFolderIsOpenedAgain() throws Exception
{
+ appendMessages(createMessages(3));
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("Message 0", "Message 1",
"Message 2");
+ context.getRouteController().startRoute("foo");
+ mock.setResultWaitTime(TimeUnit.SECONDS.toMillis(15));
+ mock.assertIsSatisfied();
+
+ assertEquals(0, countUnseen(), "every mail is consumed");
Review Comment:
Optional: the mock is satisfied as soon as "Message 2" reaches
`mock:result`. The mail is marked SEEN only later, in the `onComplete` of the
exchange. So `countUnseen()` races with that STORE. It wins today because
opening a new store takes a few round trips, but this can flake on a slow CI
machine. Waiting for it avoids that:
```suggestion
Awaitility.await().atMost(5, TimeUnit.SECONDS)
.untilAsserted(() -> assertEquals(0, countUnseen(), "every
mail is consumed"));
```
Also optional: `mock.setAssertPeriod(1000)` before `assertIsSatisfied()`
would back up the "routed once" claim in the description. Today a late
duplicate that arrives after the third mail would not fail the test.
##########
components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java:
##########
@@ -249,12 +266,16 @@ private void routeBatch(Queue<Object> exchanges, int
total) throws Exception {
// must use the original message in case we need to work around a
charset issue when extracting mail content
var msg = exchange.getIn();
final Message mail = ((MailMessage) msg).getOriginalMessage();
+ final Long uid = batchUids.get(mail);
Review Comment:
Not blocking: `batchUidValidity` is one field for the consumer, and
`rememberUids` of the next poll overwrites it. With asynchronous routing an
exchange of an earlier batch can complete after that. `mailToCommit` then
compares the current UIDVALIDITY with the value of the later batch, not the one
this UID belongs to. If UIDVALIDITY changed in between, the old UID is looked
up in the new UID space and can find another mail. That is the same failure
this PR fixes, in a rarer shape. Taking a copy here next to the UID (`final
long uidValidity = batchUidValidity;`) and passing it to `mailToCommit` closes
it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
