You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/01/07 07:24:33 UTC
[james-project] 10/10: JAMES-3687 Removes the mime message blobs
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 81223efe3d9592177c0156a901cea3d87b3a9c1c
Author: Jean Helou <jh...@codamens.fr>
AuthorDate: Sun Jan 2 23:03:06 2022 +0100
JAMES-3687 Removes the mime message blobs
- upon acknowledged dequeue
- upon successful filtering after a remove
---
.../apache/james/queue/pulsar/FilterStage.scala | 9 ++--
.../apache/james/queue/pulsar/MailMetadata.scala | 11 ++++-
.../james/queue/pulsar/PulsarMailQueue.scala | 45 +++++++++++------
.../queue/pulsar/PulsarMailQueueFactory.scala | 8 ++-
.../james/queue/pulsar/PulsarMailQueueTest.java | 57 ++++++++++++++++++++--
5 files changed, 105 insertions(+), 25 deletions(-)
diff --git a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/FilterStage.scala b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/FilterStage.scala
index 3ca8dac..534cdf1 100644
--- a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/FilterStage.scala
+++ b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/FilterStage.scala
@@ -22,10 +22,11 @@ package org.apache.james.queue.pulsar
import akka.actor.{Actor, ActorLogging, Props}
import com.sksamuel.pulsar4s.akka.streams.CommittableMessage
import com.sksamuel.pulsar4s.{ConsumerMessage, SequenceId}
+import org.apache.james.blob.api.BlobId
import scala.math.Ordered.orderingToOrdered
-private[pulsar] class FilterStage extends Actor with ActorLogging {
+private[pulsar] class FilterStage(implicit val blobIdFactory:BlobId.Factory) extends Actor with ActorLogging {
private var filters = Set.empty[Filter]
private val name = self.path.name
@@ -41,10 +42,10 @@ private[pulsar] class FilterStage extends Actor with ActorLogging {
log.debug(s"filtering mail with active filters : $filters , sequence: ${sequenceId} metadata : $metadata")
if (shouldBeFiltered(metadata, sequenceId)) {
log.debug(s"message filtered : sequence: ${sequenceId} metadata : $metadata")
- sender() ! (None, cm)
+ sender() ! (None, Some(metadata.partsId), cm)
} else {
log.debug(s"message not filtered : sequence: ${sequenceId} metadata : $metadata")
- sender() ! (Some(metadata), cm)
+ sender() ! (Some(metadata), None, cm)
}
removeExpiredFilters(sequenceId)
@@ -78,5 +79,5 @@ private[pulsar] class FilterStage extends Actor with ActorLogging {
}
private[pulsar] object FilterStage {
- def props = Props(new FilterStage())
+ def props(implicit blobIdFactory:BlobId.Factory) = Props(new FilterStage())
}
\ No newline at end of file
diff --git a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/MailMetadata.scala b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/MailMetadata.scala
index 633da40..67a4a80 100644
--- a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/MailMetadata.scala
+++ b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/MailMetadata.scala
@@ -19,6 +19,7 @@
package org.apache.james.queue.pulsar
+import org.apache.james.blob.api.BlobId
import org.apache.james.blob.mail.MimeMessagePartsId
import org.apache.james.queue.pulsar.EnqueueId.EnqueueId
import org.apache.mailet.Mail
@@ -80,4 +81,12 @@ private[pulsar] case class MailMetadata(enqueueId: String,
remoteHost: String,
perRecipientHeaders: Map[String, Iterable[Header]],
headerBlobId: String,
- bodyBlobId: String)
\ No newline at end of file
+ bodyBlobId: String){
+
+ def partsId(implicit blobIdFactory: BlobId.Factory): MimeMessagePartsId =
+ MimeMessagePartsId.builder()
+ .headerBlobId(blobIdFactory.from(headerBlobId))
+ .bodyBlobId(blobIdFactory.from(bodyBlobId))
+ .build()
+
+}
\ No newline at end of file
diff --git a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueue.scala b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueue.scala
index db0784a..15d1f33 100644
--- a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueue.scala
+++ b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueue.scala
@@ -98,6 +98,7 @@ class PulsarMailQueue(
private val enqueueBufferSize = 10
private val requeueBufferSize = 10
+ private val awaitTimeout = 10.seconds
gaugeRegistry.register(QUEUE_SIZE_METRIC_NAME_PREFIX + name, () => getSize)
private val dequeueMetrics = metricFactory.generate(DEQUEUED_METRIC_NAME_PREFIX + name.asString)
@@ -105,7 +106,8 @@ class PulsarMailQueue(
private implicit val implicitSystem: ActorSystem = system
private implicit val ec: ExecutionContextExecutor = system.dispatcher
- private implicit val client = PulsarClient(config.brokerUri)
+ private implicit val implicitBlobIdFactory: BlobId.Factory = blobIdFactory
+ private implicit val client: PulsarAsyncClient = PulsarClient(config.brokerUri)
private val admin = {
val builder = PulsarAdmin.builder()
builder.serviceHttpUrl(config.adminUri).build()
@@ -221,20 +223,18 @@ class PulsarMailQueue(
.map(message =>
(Json.fromJson[MailMetadata](Json.parse(message.message.value)).get,
message)
- ).ask[(Option[MailMetadata], CommittableMessage[String])](filterStage)
+ ).ask[(Option[MailMetadata], Option[MimeMessagePartsId], CommittableMessage[String])](filterStage)
.flatMapConcat {
- case (None, committableMessage) =>
+ case (None, Some(partsId), committableMessage) =>
committableMessage.ack()
- Source.empty
- case (Some(metadata), committableMessage) =>
- val partsId = MimeMessagePartsId.builder()
- .headerBlobId(blobIdFactory.from(metadata.headerBlobId))
- .bodyBlobId(blobIdFactory.from(metadata.bodyBlobId))
- .build()
+ deleteMimeMessage(partsId)
+ .flatMapConcat(_ => Source.empty)
+ case (Some(metadata), _, committableMessage) =>
+ val partsId = metadata.partsId
Source
.fromPublisher(readMimeMessage(partsId))
- .map(message => (readMail(metadata, message), committableMessage))
- }.map { case (mail, message) => new PulsarMailQueueItem(mail, message) }
+ .map(message => (readMail(metadata, message), partsId, committableMessage))
+ }.map { case (mail, partsId, message) => new PulsarMailQueueItem(mail, partsId, message) }
.map(mailQueueItemDecoratorFactory.decorate(_, name))
.alsoTo(counter)
@@ -251,15 +251,16 @@ class PulsarMailQueue(
.toMat(Sink.asPublisher[MailQueue.MailQueueItem](true).withAttributes(Attributes.inputBuffer(initial = 1, max = 1)))(Keep.both)
}
- class PulsarMailQueueItem(mail: Mail, message: CommittableMessage[String]) extends MailQueueItem {
+ class PulsarMailQueueItem(mail: Mail, partsId: MimeMessagePartsId, message: CommittableMessage[String]) extends MailQueueItem {
override val getMail: Mail = mail
override def done(success: Boolean): Unit = {
if (success) {
dequeueMetrics.increment()
- message.ack(cumulative = false)
+ Await.ready(message.ack(cumulative = false), awaitTimeout)
+ deleteMimeMessage(partsId).run()
} else {
- message.nack()
+ Await.ready(message.nack(), awaitTimeout)
}
}
}
@@ -333,7 +334,7 @@ class PulsarMailQueue(
private def syncEnqueue(mail: Mail, delay: Duration): Unit = {
metricFactory.decorateSupplierWithTimerMetric(
ENQUEUED_TIMER_METRIC_NAME_PREFIX + name.asString,
- () => Await.result(internalEnqueue(mail, delay), 10.second)
+ () => Await.result(internalEnqueue(mail, delay), awaitTimeout)
)
}
@@ -453,7 +454,7 @@ class PulsarMailQueue(
)
if (reader.hasMessageAvailable)
- reader.next(10.seconds)
+ reader.next(awaitTimeout)
else
None
}
@@ -584,4 +585,16 @@ class PulsarMailQueue(
throw new MailQueue.MailQueueException("Error while reading blob", e)
}
+ private def deleteMimeMessage(partsId: MimeMessagePartsId): Source[Void, NotUsed] = {
+ def doDelete() =
+ try {
+ mimeMessageStore.delete(partsId)
+ } catch {
+ case e: MessagingException =>
+ throw new MailQueue.MailQueueException("Error while deleting blob", e)
+ }
+
+ Source.fromPublisher(doDelete())
+ }
+
}
diff --git a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueueFactory.scala b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueueFactory.scala
index ca8772b..96fe42e 100644
--- a/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueueFactory.scala
+++ b/server/queue/queue-pulsar/src/main/scala/org/apache/james/queue/pulsar/PulsarMailQueueFactory.scala
@@ -51,7 +51,13 @@ class PulsarMailQueueFactory @Inject()(config: PulsarConfiguration,
private val system: ActorSystem = ActorSystem("pulsar-mailqueue")
@PreDestroy
- def stop(): Unit = system.terminate()
+ def stop(): Unit = {
+ queues.getAndUpdate(map => {
+ map.values.foreach(_.close())
+ map.empty
+ })
+ system.terminate()
+ }
override def getQueue(name: MailQueueName, count: MailQueueFactory.PrefetchCount): Optional[PulsarMailQueue] = {
Try(admin.topics().getInternalInfo(s"persistent://${config.namespace.asString}/James-${name.asString()}")).toOption.map(_ =>
diff --git a/server/queue/queue-pulsar/src/test/java/org/apache/james/queue/pulsar/PulsarMailQueueTest.java b/server/queue/queue-pulsar/src/test/java/org/apache/james/queue/pulsar/PulsarMailQueueTest.java
index 5afb06e..a992a45 100644
--- a/server/queue/queue-pulsar/src/test/java/org/apache/james/queue/pulsar/PulsarMailQueueTest.java
+++ b/server/queue/queue-pulsar/src/test/java/org/apache/james/queue/pulsar/PulsarMailQueueTest.java
@@ -21,8 +21,9 @@ package org.apache.james.queue.pulsar;
import static org.apache.james.queue.api.Mails.defaultMail;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.util.List;
+import java.util.Objects;
import java.util.stream.IntStream;
import javax.mail.MessagingException;
@@ -48,8 +49,8 @@ import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.ManageableMailQueueContract;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
-import org.apache.james.queue.pulsar.PulsarMailQueue;
import org.apache.james.server.blob.deduplication.PassThroughBlobStore;
+import org.apache.mailet.Mail;
import org.apache.mailet.base.MailAddressFixture;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
@@ -79,13 +80,14 @@ public class PulsarMailQueueTest implements MailQueueContract, MailQueueMetricCo
private MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem;
private PulsarConfiguration config;
private ActorSystem system;
+ private MemoryBlobStoreDAO memoryBlobStore;
@BeforeEach
void setUp(DockerPulsarExtension.DockerPulsar pulsar, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) {
this.metricTestSystem = metricTestSystem;
blobIdFactory = new HashBlobId.Factory();
- MemoryBlobStoreDAO memoryBlobStore = new MemoryBlobStoreDAO();
+ memoryBlobStore = new MemoryBlobStoreDAO();
PassThroughBlobStore blobStore = new PassThroughBlobStore(memoryBlobStore, BucketName.DEFAULT, blobIdFactory);
MimeMessageStore.Factory mimeMessageStoreFactory = new MimeMessageStore.Factory(blobStore);
mimeMessageStore = mimeMessageStoreFactory.mimeMessageStore();
@@ -183,6 +185,55 @@ public class PulsarMailQueueTest implements MailQueueContract, MailQueueMetricCo
.containsExactly("namez");
}
+ @Test
+ void queueShouldRemoveMailFromStoreOnAcknowledgedDequeue() throws Exception {
+ String expectedName = "name";
+ enQueue(defaultMail()
+ .name(expectedName)
+ .build());
+
+ MailQueue.MailQueueItem mailQueueItem = Flux.from(getMailQueue().deQueue()).blockFirst();
+ mailQueueItem.done(true);
+
+ assertThat(mailQueueItem.getMail().getName())
+ .isEqualTo(expectedName);
+
+ Awaitility.await().untilAsserted(() -> assertThatStoreIsEmpty());
+ }
+
+ @Test
+ void removeShouldRemoveMailFromStoreWhenFilteredOut() throws Exception {
+ enQueue(defaultMail()
+ .name("name1")
+ .build());
+ enQueue(defaultMail()
+ .name("name2")
+ .build());
+
+ getManageableMailQueue().remove(ManageableMailQueue.Type.Name, "name2");
+
+ awaitRemove();
+
+ assertThat(getManageableMailQueue().browse())
+ .toIterable()
+ .extracting(ManageableMailQueue.MailQueueItemView::getMail)
+ .extracting(Mail::getName)
+ .containsExactly("name1");
+
+ MailQueue.MailQueueItem mailQueueItem = Flux.from(getMailQueue().deQueue()).blockFirst();
+ mailQueueItem.done(true);
+ Awaitility.await().untilAsserted(() -> assertThatStoreIsEmpty());
+ }
+
+ private void assertThatStoreIsEmpty() {
+ var blobIds = Flux.from(memoryBlobStore.listBlobs(BucketName.DEFAULT))
+ .map(Objects::toString)
+ .collectList()
+ .defaultIfEmpty(List.of())
+ .block();
+ assertThat(blobIds).isEmpty();
+ }
+
@Disabled("this guarantee is too strong for Pulsar implementation and doesn't match any domain requirement")
@Override
public void flushShouldPreserveBrowseOrder() {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org