Repository: softwaremill/elasticmq (scala)

Issue:
[BUG] FIFO queue header keys set for Standard Queue as well
**Dependencies of interest:-**
org.springframework.cloud:spring-cloud-starter-sleuth:2.1.1.RELEASE
org.springframework:spring-jms:5.1.5.RELEASE
com.amazonaws:amazon-sqs-java-messaging-lib:1.0.4
image: localstack/localstack

**Error -**
While consuming a message from SQS, I am getting the below error as Elacticmq is adding these two properties - **JMSXGroupID**, **JMS_SQS_DeduplicationId** in the message even for STANDARD queue.

The problem is that these key's shouldn't be existing as my queue is standard (not FIFO). But instead, there keys are set and their values are **empty string**. Hence I am getting this exception while listening to the messages.


 [2019-05-15 04:26:44.881] [DefaultMessageListenerContainer-2] WARN  o.s.j.l.DefaultMessageListenerContainer.invokeErrorHandler(929) [demand,,] [] [] - Execution of JMS message listener failed, and no ErrorHandler has been set.
 java.lang.IllegalArgumentException: Property value can not be null or empty.
 	at com.amazon.sqs.javamessaging.message.SQSMessage.setObjectProperty(SQSMessage.java:893)
 	at brave.jms.PropertyFilter.doFilterProperties(PropertyFilter.java:60)
 	at brave.jms.PropertyFilter$1.filterProperties(PropertyFilter.java:22)
 	at brave.jms.JmsTracing.extractAndClearMessage(JmsTracing.java:176)
 	at brave.jms.JmsTracing.nextSpan(JmsTracing.java:162)
 	at org.springframework.jms.config.TracingMessagingMessageListenerAdapter.onMessage(TracingJmsListenerEndpointRegistry.java:284)
 	at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:736)
 	at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:696)
 	at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:674)
 	at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:318)
 	at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:257)
 	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1189)
 	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1179)
 	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1076)
 	at java.lang.Thread.run(Thread.java:748)

##########################
**To debug check this out ------**
##########################

_SQSMessage.java has a method to setObjectProperty. It expects the name and value to be not null and not empty._

```
@Override
    public void setObjectProperty(String name, Object value) throws JMSException {
        if (name == null || name.isEmpty()) {
            throw new IllegalArgumentException("Property name can not be null or empty.");
        }
        if (value == null || "".equals(value)) {
            throw new IllegalArgumentException("Property value can not be null or empty.");
        }
        if(!isValidPropertyValueType(value)) {
            throw new MessageFormatException("Value of property with name " + name + " has incorrect type " + value.getClass().getName() + ".");
        }
        checkPropertyWritePermissions();
        properties.put(name, new JMSMessagePropertyValue(value));
}
```

PropertyFilter.java class (of sleuth instrumentation) has this code where it clears the JMS message properties and adds them again. (JMS spec doesn't allow to add properties before clearing them!)

```
static void doFilterProperties(Message message, Set<String> namesToClear, ArrayList<Object> retainedProperties) throws JMSException {

  Enumeration<?> names = message.getPropertyNames();
  while (names.hasMoreElements()) {
    String name = (String) names.nextElement();
    Object value = message.getObjectProperty(name);
    if (!namesToClear.contains(name) && value != null) {
      retainedProperties.add(name);
      retainedProperties.add(value);
    }
  }
  message.clearProperties();
  for (int i = 0, length = retainedProperties.size(); i < length; i += 2) {
    message.setObjectProperty(
        retainedProperties.get(i).toString(),
        retainedProperties.get(i + 1)
    );
  }
}
```

**As you can see, retainedProperties adds the name value pair even if our value is empty string. It just has a non-null check. When  message.setObjectProperty(...) is executed, it will call SQSMessage.java's method and fail because of empty value of JMSXGroupID and JMS_SQS_DeduplicationId properties.**

_We should either have SQS not add these empty properties, or update the instrumentation jar to have isEmpty() check as well. The former solution should be more preferable as sleuth instrumentation class is generic and some client's might want to add empty headers._


<img width="968" alt="Screenshot 2019-05-16 at 5 40 47 AM" src="https://user-images.githubusercontent.com/50291163/57817507-4e601000-779d-11e9-9296-d884220492fb.png">

Pull request:
Fixes #204 

In case of nullified MessageDeduplicationId and MessageGroupId we should not send them at all at ReceiveMessage

Patch:
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala
index 8984c9756..d7d72cc1d 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ReceiveMessageDirectives.scala
@@ -93,8 +93,8 @@ trait ReceiveMessageDirectives {
             Rule(SenderIdAttribute, () => Some("127.0.0.1")),
             Rule(SentTimestampAttribute, () => Some(msg.created.getMillis.toString)),
             Rule(ApproximateReceiveCountAttribute, () => Some(msg.statistics.approximateReceiveCount.toString)),
-            Rule(MessageDeduplicationIdAttribute, () => msg.messageDeduplicationId.map(_.id).orElse(Some(""))),
-            Rule(MessageGroupIdAttribute, () => msg.messageGroupId.orElse(Some(""))),
+            Rule(MessageDeduplicationIdAttribute, () => msg.messageDeduplicationId.map(_.id)),
+            Rule(MessageGroupIdAttribute, () => msg.messageGroupId),
             Rule(
               ApproximateFirstReceiveTimestampAttribute,
               () =>


Test patch:
diff --git a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/ReceiveMessageAttributesTest.scala b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/ReceiveMessageAttributesTest.scala
new file mode 100644
index 000000000..240bb0053
--- /dev/null
+++ b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/ReceiveMessageAttributesTest.scala
@@ -0,0 +1,47 @@
+package org.elasticmq.rest.sqs
+
+import com.amazonaws.services.sqs.model.{CreateQueueRequest, ReceiveMessageRequest, SendMessageRequest}
+import org.scalatest.matchers.should.Matchers
+
+import scala.collection.JavaConverters.{iterableAsScalaIterableConverter, mapAsScalaMapConverter}
+
+class ReceiveMessageAttributesTest extends SqsClientServerCommunication with Matchers {
+
+  val MessageDeduplicationIdKey = "MessageDeduplicationId"
+  val MessageGroupIdKey = "MessageGroupId"
+
+  test("For normal queues, MessageDeduplicationId and MessageGroupId should not be sent with ReceiveMessage response") {
+    val queueUrl = client.createQueue(new CreateQueueRequest("testQueue1")).getQueueUrl
+    client.sendMessage(new SendMessageRequest(queueUrl, "Message 1"))
+    val messages =
+      client.receiveMessage(new ReceiveMessageRequest(queueUrl).withAttributeNames("All")).getMessages.asScala
+
+    messages should have size 1
+    val messageAttributes = messages.head.getAttributes.asScala
+    messageAttributes.get(MessageDeduplicationIdKey) should not be defined
+    messageAttributes.get(MessageGroupIdKey) should not be defined
+  }
+
+  test("For FIFO queues, MessageDeduplicationId and MessageGroupId should be sent with ReceiveMessage response") {
+    val queueUrl = client
+      .createQueue(
+        new CreateQueueRequest("testQueue1.fifo")
+          .addAttributesEntry("FifoQueue", "true")
+          .addAttributesEntry("ContentBasedDeduplication", "true")
+      )
+      .getQueueUrl
+    client.sendMessage(
+      new SendMessageRequest(queueUrl, "Message 1")
+        .withMessageDeduplicationId("123")
+        .withMessageGroupId("456")
+    )
+    val messages =
+      client.receiveMessage(new ReceiveMessageRequest(queueUrl).withAttributeNames("All")).getMessages.asScala
+
+    messages should have size 1
+    val messageAttributes = messages.head.getAttributes.asScala
+    messageAttributes.get(MessageDeduplicationIdKey) shouldBe Some("123")
+    messageAttributes.get(MessageGroupIdKey) shouldBe Some("456")
+  }
+
+}