PulsarJMSProducer: do not ovverride some system headers if already set (

#152)

pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarJMSProducer.java

@@ -192,9 +192,15 @@ private void getProducerAndSend(Destination destination, Message message) throws
       message.setObjectProperty(prop.getKey(), prop.getValue());
     }
     message.setJMSPriority(priority);
-    message.setJMSCorrelationIDAsBytes(correlationID);
-    message.setJMSType(jmsType);
-    message.setJMSReplyTo(jmsReplyTo);
+    if (message.getJMSCorrelationIDAsBytes() == null) {
+      message.setJMSCorrelationIDAsBytes(correlationID);
+    }
+    if (message.getJMSType() == null) {
+      message.setJMSType(jmsType);
+    }
+    if (message.getJMSReplyTo() == null) {
+      message.setJMSReplyTo(jmsReplyTo);
+    }
 
     if (completionListener != null) {
       producer.send(destination, message, deliveryMode, priority, timeToLive, completionListener);

pulsar-jms/src/test/java/com/datastax/oss/pulsar/jms/JMSMessageHeaderTest.java

@@ -0,0 +1,53 @@
+package com.datastax.oss.pulsar.jms;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import com.datastax.oss.pulsar.jms.utils.PulsarContainerExtension;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSProducer;
+import javax.jms.TextMessage;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+@Timeout(value = 1, unit = TimeUnit.MINUTES)
+public class JMSMessageHeaderTest {
+
+    @RegisterExtension
+    static PulsarContainerExtension pulsarContainer = new PulsarContainerExtension();
+
+    @Test
+    public void sendMessageWithHeaderReceiveJMSContext() throws Exception {
+
+        Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
+        try (PulsarConnectionFactory factory = new PulsarConnectionFactory(properties); ) {
+            try (JMSContext context = factory.createContext()) {
+                Destination destination =
+                        context.createQueue("persistent://public/default/test-" + UUID.randomUUID());
+                try (JMSConsumer consumer = context.createConsumer(destination)) {
+                    JMSProducer producer = context.createProducer();
+                    String message = "Hey JMS!";
+                    TextMessage expTextMessage = context.createTextMessage(message);
+                    expTextMessage.setJMSReplyTo(destination);
+                    expTextMessage.setJMSType("mytype");
+                    expTextMessage.setJMSCorrelationIDAsBytes(new byte[] {1, 2, 3});
+                    producer.send(destination, expTextMessage);
+                    TextMessage actTextMessage = (TextMessage) consumer.receive();
+
+                    assertNotNull(actTextMessage);
+                    assertEquals(expTextMessage.getText(), actTextMessage.getText());
+                    assertEquals(expTextMessage.getJMSReplyTo(), actTextMessage.getJMSReplyTo());
+                    assertEquals(expTextMessage.getJMSType(), actTextMessage.getJMSType());
+                    assertArrayEquals(new byte[] {1, 2, 3}, actTextMessage.getJMSCorrelationIDAsBytes());
+                }
+            }
+        }
+    }
+}