-
Notifications
You must be signed in to change notification settings - Fork 197
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement DLQ redrive actions (#987)
* StartMessageMoveTask handling * add some logging, test cases * implement CancelMessageMoveTask * fix missing ApproximateNumberOfMessagesMoved refactor test * ListMessageMoveTasksDirectives and error handling * fix scala 2.12 build * refactor ListMessageMoveTasksDirectives * refactor QueueManagerActor * refactor TaskId => TaskHandle * make MessageMoveTaskTest support both SDKs * fix build * refactoring
- Loading branch information
Showing
31 changed files
with
1,252 additions
and
36 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,32 +1,35 @@ | ||
| package org.elasticmq | ||
| import org.elasticmq.msg.MessageMoveTaskHandle | ||
|
|
||
| trait ElasticMQError { | ||
| sealed trait ElasticMQError { | ||
| val queueName: String | ||
| val code: String | ||
| val code: String // TODO: code should be handled in rest-sqs module | ||
| val message: String | ||
| } | ||
|
|
||
| class QueueAlreadyExists(val queueName: String) extends ElasticMQError { | ||
| final case class QueueAlreadyExists(val queueName: String) extends ElasticMQError { | ||
| val code = "QueueAlreadyExists" | ||
| val message = s"Queue already exists: $queueName" | ||
| } | ||
|
|
||
| case class QueueCreationError(queueName: String, reason: String) extends ElasticMQError { | ||
| val code = "QueueCreationError" | ||
| val message = s"Queue named $queueName could not be created because of $reason" | ||
| } | ||
|
|
||
| case class InvalidParameterValue(queueName: String, reason: String) extends ElasticMQError { | ||
| final case class InvalidParameterValue(queueName: String, reason: String) extends ElasticMQError { | ||
| val code = "InvalidParameterValue" | ||
| val message = reason | ||
| } | ||
|
|
||
| class MessageDoesNotExist(val queueName: String, messageId: MessageId) extends ElasticMQError { | ||
| val code = "MessageDoesNotExist" | ||
| val message = s"Message does not exist: $messageId in queue: $queueName" | ||
| } | ||
|
|
||
| class InvalidReceiptHandle(val queueName: String, receiptHandle: String) extends ElasticMQError { | ||
| final case class InvalidReceiptHandle(val queueName: String, receiptHandle: String) extends ElasticMQError { | ||
| val code = "ReceiptHandleIsInvalid" | ||
| val message = s"""The receipt handle "$receiptHandle" is not valid.""" | ||
| } | ||
|
|
||
| final case class InvalidMessageMoveTaskHandle(val taskHandle: MessageMoveTaskHandle) extends ElasticMQError { | ||
| val code = "ResourceNotFoundException" | ||
| val message = s"""The task handle "$taskHandle" is not valid or does not exist""" | ||
|
|
||
| override val queueName: String = "invalid" | ||
| } | ||
|
|
||
| final case class MessageMoveTaskAlreadyRunning(val queueName: String) extends ElasticMQError { | ||
| val code = "AWS.SimpleQueueService.UnsupportedOperation" | ||
| val message = s"""A message move task is already running on queue "$queueName"""" | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 19 additions & 0 deletions
19
core/src/main/scala/org/elasticmq/actor/QueueManagerActorStorage.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| package org.elasticmq.actor | ||
| import org.apache.pekko.actor.{ActorContext, ActorRef} | ||
| import org.apache.pekko.util.Timeout | ||
| import org.elasticmq.QueueData | ||
|
|
||
| import scala.collection.mutable | ||
| import scala.concurrent.ExecutionContext | ||
| import scala.concurrent.duration.DurationInt | ||
|
|
||
| trait QueueManagerActorStorage { | ||
|
|
||
| def context: ActorContext | ||
|
|
||
| implicit lazy val ec: ExecutionContext = context.dispatcher | ||
| implicit lazy val timeout: Timeout = 5.seconds | ||
|
|
||
| case class ActorWithQueueData(actorRef: ActorRef, queueData: QueueData) | ||
| def queues: mutable.Map[String, ActorWithQueueData] | ||
| } |
92 changes: 92 additions & 0 deletions
92
core/src/main/scala/org/elasticmq/actor/QueueManagerMessageMoveOps.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| package org.elasticmq.actor | ||
| import org.apache.pekko.actor.ActorRef | ||
| import org.apache.pekko.util.Timeout | ||
| import org.elasticmq.actor.reply._ | ||
| import org.elasticmq.msg.{CancelMovingMessages, GetQueueData, MessageMoveTaskHandle, StartMovingMessages} | ||
| import org.elasticmq.util.Logging | ||
| import org.elasticmq.{ElasticMQError, InvalidMessageMoveTaskHandle} | ||
|
|
||
| import scala.collection.mutable | ||
| import scala.concurrent.Future | ||
| import scala.util.{Failure, Success} | ||
|
|
||
| trait QueueManagerMessageMoveOps extends Logging { | ||
| this: QueueManagerActorStorage => | ||
|
|
||
| private val messageMoveTasks = mutable.HashMap[MessageMoveTaskHandle, ActorRef]() | ||
|
|
||
| def startMessageMoveTask( | ||
| sourceQueue: ActorRef, | ||
| sourceArn: String, | ||
| destinationQueue: Option[ActorRef], | ||
| destinationArn: Option[String], | ||
| maxNumberOfMessagesPerSecond: Option[Int] | ||
| )(implicit timeout: Timeout): ReplyAction[Either[ElasticMQError, MessageMoveTaskHandle]] = { | ||
| val self = context.self | ||
| val replyTo = context.sender() | ||
| (for { | ||
| destinationQueueActorRef <- destinationQueue | ||
| .map(Future.successful) | ||
| .getOrElse(findDeadLetterQueueSource(sourceQueue)) | ||
| result <- sourceQueue ? StartMovingMessages( | ||
| destinationQueueActorRef, | ||
| destinationArn, | ||
| sourceArn, | ||
| maxNumberOfMessagesPerSecond, | ||
| self | ||
| ) | ||
| } yield (result, destinationQueueActorRef)).onComplete { | ||
| case Success((result, destinationQueueActorRef)) => | ||
| result match { | ||
| case Right(taskHandle) => | ||
| logger.debug("Message move task {} => {} created", sourceQueue, destinationQueueActorRef) | ||
| messageMoveTasks.put(taskHandle, sourceQueue) | ||
| replyTo ! Right(taskHandle) | ||
| case Left(error) => | ||
| logger.error("Failed to start message move task: {}", error) | ||
| replyTo ! Left(error) | ||
| } | ||
| case Failure(ex) => logger.error("Failed to start message move task", ex) | ||
| } | ||
| DoNotReply() | ||
| } | ||
|
|
||
| def onMessageMoveTaskFinished(taskHandle: MessageMoveTaskHandle): ReplyAction[Unit] = { | ||
| logger.debug("Message move task {} finished", taskHandle) | ||
| messageMoveTasks.remove(taskHandle) | ||
| DoNotReply() | ||
| } | ||
|
|
||
| def cancelMessageMoveTask(taskHandle: MessageMoveTaskHandle): ReplyAction[Either[ElasticMQError, Long]] = { | ||
| logger.info("Cancelling message move task {}", taskHandle) | ||
| messageMoveTasks.get(taskHandle) match { | ||
| case Some(sourceQueue) => | ||
| val replyTo = context.sender() | ||
| sourceQueue ? CancelMovingMessages() onComplete { | ||
| case Success(numMessageMoved) => | ||
| logger.debug("Message move task {} cancelled", taskHandle) | ||
| messageMoveTasks.remove(taskHandle) | ||
| replyTo ! Right(numMessageMoved) | ||
| case Failure(ex) => | ||
| logger.error("Failed to cancel message move task", ex) | ||
| replyTo ! Left(ex) | ||
| } | ||
| DoNotReply() | ||
| case None => | ||
| ReplyWith(Left(new InvalidMessageMoveTaskHandle(taskHandle))) | ||
| } | ||
| } | ||
|
|
||
| private def findDeadLetterQueueSource(sourceQueue: ActorRef) = { | ||
| val queueDataF = sourceQueue ? GetQueueData() | ||
| queueDataF.map { queueData => | ||
| queues | ||
| .filter { case (_, data) => | ||
| data.queueData.deadLettersQueue.exists(dlqData => dlqData.name == queueData.name) | ||
| } | ||
| .head | ||
| ._2 | ||
| .actorRef | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.