This is an automated email from the ASF dual-hosted git repository.
markap14 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new f73888e7dd NIFI-12478 Return Message Type as body for JMS Object
Messages (#8131)
f73888e7dd is described below
commit f73888e7dd4b0df37f5845bc318a7ad85f22d75d
Author: David Handermann <[email protected]>
AuthorDate: Thu Dec 7 15:41:34 2023 -0600
NIFI-12478 Return Message Type as body for JMS Object Messages (#8131)
---
.../java/org/apache/nifi/jms/processors/JMSConsumer.java | 4 +++-
.../nifi/jms/processors/MessageBodyToBytesConverter.java | 15 ---------------
.../nifi/jms/processors/JMSPublisherConsumerIT.java | 13 ++++++-------
pom.xml | 1 -
4 files changed, 9 insertions(+), 24 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
index 3c721764ad..d39e625d7a 100644
---
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
+++
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
@@ -38,6 +38,7 @@ import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import javax.jms.Topic;
import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
@@ -184,7 +185,8 @@ class JMSConsumer extends JMSWorker {
messageBody =
MessageBodyToBytesConverter.toBytes((BytesMessage) message);
} else if (message instanceof ObjectMessage) {
messageType = ObjectMessage.class.getSimpleName();
- messageBody =
MessageBodyToBytesConverter.toBytes((ObjectMessage) message);
+ // Return Message Type as body to avoid unsupported class
references
+ messageBody = messageType.getBytes(StandardCharsets.UTF_8);
} else if (message instanceof StreamMessage) {
messageType = StreamMessage.class.getSimpleName();
messageBody =
MessageBodyToBytesConverter.toBytes((StreamMessage) message);
diff --git
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/MessageBodyToBytesConverter.java
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/MessageBodyToBytesConverter.java
index c8cd9adbc9..e6c0d3e9e4 100644
---
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/MessageBodyToBytesConverter.java
+++
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/MessageBodyToBytesConverter.java
@@ -31,13 +31,11 @@ import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageEOFException;
-import javax.jms.ObjectMessage;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.SerializationUtils;
/**
*
@@ -88,19 +86,6 @@ abstract class MessageBodyToBytesConverter {
}
}
- /**
- *
- * @param message instance of {@link ObjectMessage}
- * @return byte array representing the {@link ObjectMessage}
- */
- public static byte[] toBytes(ObjectMessage message) {
- try {
- return SerializationUtils.serialize(message.getObject());
- } catch (Exception e) {
- throw new MessageConversionException("Failed to convert " +
ObjectMessage.class.getSimpleName() + " to byte[]", e);
- }
- }
-
/**
* @param message instance of {@link StreamMessage}
* @return byte array representing the {@link StreamMessage}
diff --git
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
index 36dd3828e7..a2d9f6e2f1 100644
---
a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
+++
b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/test/java/org/apache/nifi/jms/processors/JMSPublisherConsumerIT.java
@@ -18,7 +18,6 @@ package org.apache.nifi.jms.processors;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.commons.lang3.SerializationUtils;
import org.apache.nifi.jms.processors.JMSConsumer.JMSResponse;
import org.apache.nifi.logging.ComponentLog;
import org.junit.jupiter.api.Test;
@@ -70,9 +69,9 @@ public class JMSPublisherConsumerIT {
};
Consumer<JMSResponse> responseChecker = response -> {
- assertEquals(
- "stringAsObject",
- SerializationUtils.deserialize(response.getMessageBody())
+ assertArrayEquals(
+
ObjectMessage.class.getSimpleName().getBytes(StandardCharsets.UTF_8),
+ response.getMessageBody()
);
};
@@ -447,7 +446,7 @@ public class JMSPublisherConsumerIT {
}
callbackInvoked.set(true);
- assertEquals("1", new String(response.getMessageBody()));
+ assertEquals("2", new String(response.getMessageBody()));
acknowledge(response);
});
}
@@ -464,7 +463,7 @@ public class JMSPublisherConsumerIT {
}
callbackInvoked.set(true);
- assertEquals("2", new
String(response.getMessageBody()));
+ assertEquals("1", new
String(response.getMessageBody()));
throw new RuntimeException("intentional to avoid
explicit ack");
});
}
@@ -483,7 +482,7 @@ public class JMSPublisherConsumerIT {
}
callbackInvoked.set(true);
- assertEquals("2", new
String(response.getMessageBody()));
+ assertEquals("1", new
String(response.getMessageBody()));
acknowledge(response);
});
}
diff --git a/pom.xml b/pom.xml
index ef173cb3a0..edc9664bcc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1166,7 +1166,6 @@
!ITDeleteAzureBlobStorage_v12,
!ITMoveAzureDataLakeStorage,
!AzureGraphUserGroupProviderIT,
-
!JMSPublisherConsumerIT#validateMessageRedeliveryWhenNotAcked,
!GremlinClientServiceYamlSettingsAndBytecodeIT,
!GremlinClientServiceControllerSettingsIT,
!ITestConsumeEmail#validateUrl,
