Repository: softwaremill/elasticmq (scala)

Issue:
SentTimestamp & ApproximateFirstReceiveTimestamp for the message not carried over to DLQ
The SentTimestamp & ApproximateFirstReceiveTimestamp are not carried over from one SQS to its dead letter queue (DLQ)

We have a scenario in our app where we depend on original SentTimestamp and this came out when I was running my tests using this lib instead of real AWS SQS.

Flow:
Create 2 queues:
- Primary
- PrimaryDLQ (this is the dead letter queue (DLQ) for the above Primary queue)
Set the RedrivePolicy on Primary with maxReceiveCount=2

After receiving the message twice from the Primary queue, the message is moved to PrimaryDLQ, however, this elasticmq is not carrying over the original SentTimestamp & ApproximateFirstReceiveTimestamp which was set when the message came to the Primary queue. It added the timestamp on which the message came to the PrimaryDLQ queue.

AWS SQS carries the same SentTimestamp & ApproximateFirstReceiveTimestamp from the original message over to DLQ. I tested this on AWS SQS creating 2 SQS and performing the same test as stated above.


AWS SQS - 
Message Receive - 1
aws sqs receive-message --queue-url https://sqs.us-east-1.amazonaws.com/*/narayar-local-retry --attribute-names All --message-attribute-names All
{
    "Messages": [
        {
            "Body": "foo bar",
            "Attributes": {
                "ApproximateFirstReceiveTimestamp": "1599031260287",
                "ApproximateReceiveCount": "1",
                "SentTimestamp": "1599030599229"
            },
            "MD5OfBody": "327b6f07435811239bc47e1544353273",
            "MessageId": "938bd0cf-15ab-4115-8396-83e48ba17f27"
        }
    ]
}


Message Receive - 2
aws sqs receive-message --queue-url https://sqs.us-east-1.amazonaws.com/*/narayar-local-retry --attribute-names All --message-attribute-names All
{
    "Messages": [
        {
            "Body": "foo bar",
            "Attributes": {
                "ApproximateFirstReceiveTimestamp": "1599031260287",
                "ApproximateReceiveCount": "2",
                "SentTimestamp": "1599030599229"
            },
            "MD5OfBody": "327b6f07435811239bc47e1544353273",
            "MessageId": "938bd0cf-15ab-4115-8396-83e48ba17f27"
        }
    ]
}

Message moved to DLQ - Message received from DLQ
aws sqs receive-message --queue-url https://sqs.us-east-1.amazonaws.com/*/narayar-local-dlq --attribute-names All --message-attribute-names All
{
    "Messages": [
        {
            "Body": "foo bar",
            "Attributes": {
                "ApproximateFirstReceiveTimestamp": "1599031260287",
                "ApproximateReceiveCount": "4",
                "SentTimestamp": "1599030599229"
            },
            "MD5OfBody": "327b6f07435811239bc47e1544353273",
            "MessageId": "938bd0cf-15ab-4115-8396-83e48ba17f27"
        }
    ]
}


elasticmq - 
Message Receive - 1
aws --endpoint-url=http://localhost:9324 sqs receive-message --queue-url http://localhost:9324/queue/WebhooksmsosxRetryQueue --attribute-names All --message-attribute-names All
{
    "Messages": [
        {
            "Body": "Message 111",
            "Attributes": {
                "ApproximateFirstReceiveTimestamp": "1599031761855",
                "SenderId": "127.0.0.1",
                "MessageDeduplicationId": "",
                "SentTimestamp": "1599031742818",
                "ApproximateReceiveCount": "1",
                "MessageGroupId": ""
            },
            "ReceiptHandle": "d1a5ef43-264f-4a68-8253-73999189de44#731d8e04-7093-4c23-a2f8-4da5577079b9",
            "MD5OfBody": "f6d67b966ed9d4c93492f3573c9b0d14",
            "MessageId": "d1a5ef43-264f-4a68-8253-73999189de44"
        }
    ]
}

Message Receive - 2
aws --endpoint-url=http://localhost:9324 sqs receive-message --queue-url http://localhost:9324/queue/WebhooksmsosxRetryQueue --attribute-names All --message-attribute-names All
{
    "Messages": [
        {
            "Body": "Message 111",
            "Attributes": {
                "ApproximateFirstReceiveTimestamp": "1599031761855",
                "SenderId": "127.0.0.1",
                "MessageDeduplicationId": "",
                "SentTimestamp": "1599031742818",
                "ApproximateReceiveCount": "2",
                "MessageGroupId": ""
            },
            "ReceiptHandle": "d1a5ef43-264f-4a68-8253-73999189de44#a3f71654-e6f4-4522-9d3a-1af51a9bc491",
            "MD5OfBody": "f6d67b966ed9d4c93492f3573c9b0d14",
            "MessageId": "d1a5ef43-264f-4a68-8253-73999189de44"
        }
    ]
}

Message moved to DLQ - Message received from DLQ
aws --endpoint-url=http://localhost:9324 sqs receive-message --queue-url http://localhost:9324/queue/WebhooksmsosxDLQ --attribute-names All --message-attribute-names All
{
    "Messages": [
        {
            "Body": "Message 111",
            "Attributes": {
                "ApproximateFirstReceiveTimestamp": "1599031846325",
                "SenderId": "127.0.0.1",
                "MessageDeduplicationId": "",
                "SentTimestamp": "1599031833907",
                "ApproximateReceiveCount": "1",
                "MessageGroupId": ""
            },
            "ReceiptHandle": "d1a5ef43-264f-4a68-8253-73999189de44#01afda03-68f0-4ceb-86b3-ba41723687b1",
            "MD5OfBody": "f6d67b966ed9d4c93492f3573c9b0d14",
            "MessageId": "d1a5ef43-264f-4a68-8253-73999189de44"
        }
    ]
}

Pull request:
Fixes #386.

In real SQS, when message is moved to DLQ the SentTimestamp and ApproximateFirstReceiveTimestamp are not preserved.
In ElasticMQ, both of attributes were initialised to new values. Once we were moving message to DLQ, we were treating it as completely new message, thus we were initialising attributes to default values.

Fix assumes, that message that is being sent to DLQ is not a new message. We are treating it as already existing one with filled values that should be copied to DLQ.

Patch:
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
index 0a225c965..5f8070810 100644
--- a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
+++ b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorMessageOps.scala
@@ -22,6 +22,7 @@ trait QueueActorMessageOps extends Logging {
         receiveMessages(visibilityTimeout, count, receiveRequestAttemptId)
       case DeleteMessage(deliveryReceipt) => deleteMessage(deliveryReceipt)
       case LookupMessage(messageId)       => messageQueue.byId.get(messageId.id).map(_.toMessageData)
+      case MoveMessage(message)           => moveMessage(message)
     }
 
   private def handleOrRedirectMessage(message: NewMessageData): ReplyAction[MessageData] = {
@@ -123,7 +124,7 @@ trait QueueActorMessageOps extends Logging {
           case default => Some(default)
         }
       })
-      .getOrElse(getMessagesFromQueue(visibilityTimeout, count))
+      .getOrElse(getMessagesFromQueue(count))
       .map { internalMessage =>
         // Putting the msg again into the queue, with a new next delivery
         val newNextDelivery = computeNextDelivery(visibilityTimeout)
@@ -150,12 +151,12 @@ trait QueueActorMessageOps extends Logging {
     }
   }
 
-  private def getMessagesFromQueue(visibilityTimeout: VisibilityTimeout, count: Int) = {
+  private def getMessagesFromQueue(count: Int) = {
     val deliveryTime = nowProvider.nowMillis
     messageQueue.dequeue(count, deliveryTime).flatMap { internalMessage =>
       if (queueData.deadLettersQueue.map(_.maxReceiveCount).exists(_ <= internalMessage.receiveCount)) {
         logger.debug(s"${queueData.name}: send message $internalMessage to dead letters actor $deadLettersActorRef")
-        deadLettersActorRef.foreach(_ ! SendMessage(internalMessage.toNewMessageData))
+        deadLettersActorRef.foreach(_ ! MoveMessage(internalMessage))
         internalMessage.deliveryReceipts.foreach(dr => deleteMessage(DeliveryReceipt(dr)))
         None
       } else {
@@ -185,4 +186,27 @@ trait QueueActorMessageOps extends Logging {
       }
     }
   }
+
+  private def moveMessage(message: InternalMessage): Unit = {
+    def moveMessageToQueue(internalMessage: InternalMessage): Unit = {
+      messageQueue += internalMessage
+      logger.debug(s"Moved message with id ${internalMessage.id} to DLQ ${queueData.name}")
+    }
+
+    copyMessagesToActorRef.foreach { _ ! SendMessage(message.toNewMessageData) }
+
+    if (queueData.isFifo) {
+      // Ensure a message with the same deduplication id is not on the queue already. If the message is already on the
+      // queue do nothing.
+      // TODO: A message dedup id should be checked up to 5 mins after it has been received. If it has been deleted
+      // during that period, it should _still_ be used when deduplicating new messages. If there's a match with a
+      // deleted message (that was sent less than 5 minutes ago, the new message should not be added).
+      messageQueue.byId.values.find(isDuplicate(message.toNewMessageData, _)) match {
+        case Some(_) => ()
+        case None    => moveMessageToQueue(message)
+      }
+    } else {
+      moveMessageToQueue(message)
+    }
+  }
 }
diff --git a/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala b/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala
index 962849780..037df5ae1 100644
--- a/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala
+++ b/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala
@@ -2,6 +2,7 @@ package org.elasticmq.msg
 
 import akka.actor.ActorRef
 import org.elasticmq._
+import org.elasticmq.actor.queue.InternalMessage
 import org.elasticmq.actor.reply.Replyable
 import org.joda.time.Duration
 
@@ -25,6 +26,7 @@ case class GetQueueStatistics(deliveryTime: Long) extends QueueQueueMsg[QueueSta
 case class ClearQueue() extends QueueQueueMsg[Unit]
 
 case class SendMessage(message: NewMessageData) extends QueueMessageMsg[MessageData]
+case class MoveMessage(message: InternalMessage) extends QueueMessageMsg[Unit]
 case class UpdateVisibilityTimeout(messageId: MessageId, visibilityTimeout: VisibilityTimeout)
     extends QueueMessageMsg[Either[MessageDoesNotExist, Unit]]
 case class ReceiveMessages(


Test patch:
diff --git a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
index 395a12783..4453f884b 100644
--- a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
+++ b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
@@ -1741,6 +1741,66 @@ class AmazonJavaSdkTestSuite extends AnyFunSuite with Matchers with BeforeAndAft
     newQueueAttributes(redrivePolicyAttribute) should be(redrivePolicy)
   }
 
+  test(
+    "When message is moved to dead letter queue, its SentTimestamp and ApproximateFirstReceiveTimestamp attributes should not be changed"
+  ) {
+    import org.elasticmq.rest.sqs.model.RedrivePolicyJson._
+    import spray.json._
+
+    def getOneMessage(queueUrl: String) = {
+      client
+        .receiveMessage(new ReceiveMessageRequest(queueUrl).withWaitTimeSeconds(2).withAttributeNames("All"))
+        .getMessages
+        .asScala
+        .headOption
+        .orNull
+    }
+
+    // Given
+    val deadLetterQueueUrl = client.createQueue(new CreateQueueRequest("dlq1")).getQueueUrl
+    val redrivePolicy = RedrivePolicy("dlq1", awsRegion, awsAccountId, 2).toJson.toString()
+    val createQueueResult = client.createQueue(
+      new CreateQueueRequest("q1")
+        .withAttributes(Map(defaultVisibilityTimeoutAttribute -> "1", redrivePolicyAttribute -> redrivePolicy).asJava)
+    )
+
+    // When
+
+    client.sendMessage(new SendMessageRequest(createQueueResult.getQueueUrl, "test message"))
+
+    val msg1 = getOneMessage(createQueueResult.getQueueUrl)
+    val msg2 = getOneMessage(createQueueResult.getQueueUrl)
+    val emptyMsg = getOneMessage(createQueueResult.getQueueUrl)
+
+    val msgFromDeadLetterQueue = getOneMessage(deadLetterQueueUrl)
+
+    // Then
+    val msg1SentTimestamp = msg1.getAttributes.get("SentTimestamp")
+    val msg1ApproximateReceiveCount = msg1.getAttributes.get("ApproximateReceiveCount")
+    val msg1ApproximateFirstReceiveTimestamp = msg1.getAttributes.get("SentTimestamp")
+
+    val msg2SentTimestamp = msg2.getAttributes.get("SentTimestamp")
+    val msg2ApproximateReceiveCount = msg2.getAttributes.get("ApproximateReceiveCount")
+    val msg2ApproximateFirstReceiveTimestamp = msg2.getAttributes.get("SentTimestamp")
+
+    val msgFromDeadLetterQueueSentTimestamp = msgFromDeadLetterQueue.getAttributes.get("SentTimestamp")
+    val msgFromDeadLetterQueueApproximateReceiveCount =
+      msgFromDeadLetterQueue.getAttributes.get("ApproximateReceiveCount")
+    val msgFromDeadLetterQueueApproximateFirstReceiveTimestamp =
+      msgFromDeadLetterQueue.getAttributes.get("SentTimestamp")
+
+    msg1SentTimestamp shouldBe msg2SentTimestamp
+    msg2SentTimestamp shouldBe msgFromDeadLetterQueueSentTimestamp
+
+    msg2ApproximateReceiveCount.toInt shouldBe (msg1ApproximateReceiveCount.toInt + 1)
+    msgFromDeadLetterQueueApproximateReceiveCount.toInt shouldBe (msg2ApproximateReceiveCount.toInt + 1)
+
+    msg1ApproximateFirstReceiveTimestamp shouldBe msg2ApproximateFirstReceiveTimestamp
+    msg2ApproximateFirstReceiveTimestamp shouldBe msgFromDeadLetterQueueApproximateFirstReceiveTimestamp
+
+    emptyMsg shouldBe null
+  }
+
   test("should return an error when the deadletter queue does not exist") {
     import org.elasticmq.rest.sqs.model.RedrivePolicyJson._
     import spray.json._