publish =
() -> {
producer.send(msg, topicOrQueue);
diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java
index e6cd35b63b45..15fe06103fbd 100644
--- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java
+++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java
@@ -18,12 +18,19 @@
package org.apache.beam.sdk.io.solace.data;
import com.google.auto.value.AutoValue;
+import com.solacesystems.jcsmp.BytesMessage;
import com.solacesystems.jcsmp.BytesXMLMessage;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
+import com.solacesystems.jcsmp.JCSMPFactory;
+import com.solacesystems.jcsmp.TextMessage;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
import org.apache.beam.sdk.schemas.AutoValueSchema;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
+import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -112,6 +119,16 @@ public abstract static class Builder {
@AutoValue
@DefaultSchema(AutoValueSchema.class)
public abstract static class Record {
+ /** Identifies how the record payload is represented in a JCSMP message. */
+ public enum PayloadType {
+ /** The legacy XML-data payload written with {@code BytesXMLMessage.writeBytes}. */
+ BYTES_XML,
+ /** A text payload written with {@code TextMessage.setText}. */
+ TEXT,
+ /** A binary payload written with {@code BytesMessage.setData}. */
+ BYTES;
+ }
+
/**
* Gets the unique identifier of the message, a string for an application-specific message
* identifier.
@@ -255,13 +272,27 @@ public abstract static class Record {
@SchemaFieldNumber("12")
public abstract byte[] getAttachmentBytes();
+ /** Gets the JCSMP payload representation used for this record. */
+ @SchemaFieldNumber("13")
+ public abstract PayloadType getPayloadType();
+
+ /** Gets the payload decoded as UTF-8 when this record has type {@link PayloadType#TEXT}. */
+ public final String getText() {
+ if (getPayloadType() != PayloadType.TEXT) {
+ throw new IllegalStateException(
+ "Text is only available for records with payload type TEXT.");
+ }
+ return decodeUtf8(getPayload());
+ }
+
public static Builder builder() {
return new AutoValue_Solace_Record.Builder()
.setExpiration(0L)
.setPriority(-1)
.setRedelivered(false)
.setTimeToLive(0)
- .setAttachmentBytes(new byte[0]);
+ .setAttachmentBytes(new byte[0])
+ .setPayloadType(PayloadType.BYTES_XML);
}
@AutoValue.Builder
@@ -270,6 +301,14 @@ public abstract static class Builder {
public abstract Builder setPayload(byte[] payload);
+ public abstract Builder setPayloadType(PayloadType payloadType);
+
+ /** Sets a UTF-8 text payload and selects {@link PayloadType#TEXT}. */
+ public Builder setText(String text) {
+ byte[] payload = text == null ? new byte[0] : text.getBytes(StandardCharsets.UTF_8);
+ return setPayloadType(PayloadType.TEXT).setPayload(payload);
+ }
+
public abstract Builder setDestination(@Nullable Destination destination);
public abstract Builder setExpiration(long expiration);
@@ -295,6 +334,19 @@ public abstract Builder setReplicationGroupMessageId(
public abstract Record build();
}
+
+ private static String decodeUtf8(byte[] payload) {
+ try {
+ return StandardCharsets.UTF_8
+ .newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT)
+ .decode(ByteBuffer.wrap(payload))
+ .toString();
+ } catch (CharacterCodingException e) {
+ throw new IllegalArgumentException("Text payload is not valid UTF-8.", e);
+ }
+ }
}
/**
@@ -387,6 +439,7 @@ public abstract static class Builder {
*/
public static class SolaceRecordMapper {
private static final Logger LOG = LoggerFactory.getLogger(SolaceRecordMapper.class);
+
/**
* Maps a {@link BytesXMLMessage} (if not null) to a {@link Solace.Record}.
*
@@ -396,35 +449,17 @@ public static class SolaceRecordMapper {
* @param msg The Solace message to map.
* @return A Solace Record representing the message, or null if the input message was null.
*/
- public static @Nullable Record map(@Nullable BytesXMLMessage msg) {
+ public static @Nullable Record toRecord(@Nullable BytesXMLMessage msg) {
if (msg == null) {
return null;
}
- ByteArrayOutputStream payloadBytesStream = new ByteArrayOutputStream();
- if (msg.getContentLength() != 0) {
- try {
- payloadBytesStream.write(msg.getBytes());
- } catch (IOException e) {
- LOG.error("Could not write bytes from the BytesXMLMessage to the Solace.record.", e);
- }
- }
-
- ByteArrayOutputStream attachmentBytesStream = new ByteArrayOutputStream();
- if (msg.getAttachmentContentLength() != 0) {
- try {
- attachmentBytesStream.write(msg.getAttachmentByteBuffer().array());
- } catch (IOException e) {
- LOG.error(
- "Could not AttachmentByteBuffer from the BytesXMLMessage to the Solace.record.", e);
- }
- }
-
Destination replyTo = getDestination(msg.getCorrelationId(), msg.getReplyTo());
Destination destination = getDestination(msg.getCorrelationId(), msg.getDestination());
- return Record.builder()
+
+ Record.Builder recordBuilder = decodePayload(msg);
+ return recordBuilder
.setMessageId(msg.getApplicationMessageId())
- .setPayload(payloadBytesStream.toByteArray())
.setDestination(destination)
.setExpiration(msg.getExpiration())
.setPriority(msg.getPriority())
@@ -438,7 +473,6 @@ public static class SolaceRecordMapper {
msg.getReplicationGroupMessageId() != null
? msg.getReplicationGroupMessageId().toString()
: null)
- .setAttachmentBytes(attachmentBytesStream.toByteArray())
.build();
}
@@ -462,5 +496,108 @@ public static class SolaceRecordMapper {
}
return destinationBuilder.build();
}
+
+ /**
+ * Maps a {@link Record} to a {@link BytesXMLMessage}.
+ *
+ * Only the fields common to both a {@link Record} and a {@link BytesXMLMessage} are set: the
+ * payload (according to the record's {@link Record.PayloadType}), the sender timestamp
+ * (defaulting to the current time when the record does not provide one) and the application
+ * message id. Publishing-specific fields such as delivery mode or correlation key are not
+ * handled here and must be set by the caller.
+ *
+ * @param record the {@link Record} to map.
+ * @return a JCSMP {@link BytesXMLMessage} carrying the record's common fields.
+ */
+ public static BytesXMLMessage toMessage(Record record) {
+ BytesXMLMessage msg = encodePayload(record);
+
+ Long senderTimestamp = record.getSenderTimestamp();
+ if (senderTimestamp == null) {
+ senderTimestamp = System.currentTimeMillis();
+ }
+ msg.setSenderTimestamp(senderTimestamp);
+ msg.setApplicationMessageId(record.getMessageId());
+
+ return msg;
+ }
+
+ /**
+ * Reads the payload from a {@link Solace.Record} into a partially-populated {@link
+ * BytesXMLMessage}.
+ *
+ * @param record the Solace record.
+ * @return a {@link BytesXMLMessage} with the payload set based on the record's payload type.
+ */
+ private static BytesXMLMessage encodePayload(Record record) {
+ switch (record.getPayloadType()) {
+ case TEXT:
+ TextMessage text = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);
+ text.setText(record.getText());
+ return text;
+ case BYTES:
+ BytesMessage bytes = JCSMPFactory.onlyInstance().createMessage(BytesMessage.class);
+ bytes.setData(record.getPayload());
+ return bytes;
+ case BYTES_XML:
+ BytesXMLMessage xml = JCSMPFactory.onlyInstance().createBytesXMLMessage();
+ xml.writeBytes(record.getPayload());
+ if (record.getAttachmentBytes().length != 0) {
+ xml.writeAttachment(record.getAttachmentBytes());
+ }
+ return xml;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported payload type: " + record.getPayloadType());
+ }
+ }
+
+ /**
+ * Reads the payload from a {@link BytesXMLMessage} into a partially-populated {@link
+ * Record.Builder}.
+ *
+ * @param msg the JCSMP message.
+ * @return a {@link Record.Builder} with the payload and payload type set based on the message
+ * type.
+ */
+ private static Record.Builder decodePayload(@NonNull BytesXMLMessage msg) {
+ if (msg instanceof TextMessage) {
+ String text = ((TextMessage) msg).getText();
+ byte[] payload = text == null ? new byte[0] : text.getBytes(StandardCharsets.UTF_8);
+ return Record.builder().setPayloadType(Record.PayloadType.TEXT).setPayload(payload);
+ }
+
+ if (msg instanceof BytesMessage) {
+ byte[] data = ((BytesMessage) msg).getData();
+ byte[] payload = data == null ? new byte[0] : data;
+ return Record.builder().setPayloadType(Record.PayloadType.BYTES).setPayload(payload);
+ }
+
+ // BYTES_XML fallback
+ byte[] payload = readBytes(msg);
+ byte[] attachment = readAttachment(msg);
+ return Record.builder()
+ .setPayloadType(Record.PayloadType.BYTES_XML)
+ .setPayload(payload)
+ .setAttachmentBytes(attachment);
+ }
+
+ private static byte[] readBytes(BytesXMLMessage msg) {
+ if (msg.getContentLength() == 0) {
+ return new byte[0];
+ }
+ return Arrays.copyOf(msg.getBytes(), msg.getContentLength());
+ }
+
+ private static byte[] readAttachment(BytesXMLMessage msg) {
+ if (msg.getAttachmentContentLength() == 0) {
+ return new byte[0];
+ }
+
+ ByteBuffer buffer = msg.getAttachmentByteBuffer();
+ byte[] attachment = new byte[buffer.remaining()];
+ buffer.get(attachment);
+ return attachment;
+ }
}
}
diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOWriteTest.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOWriteTest.java
index e92657c3c3d2..3cdc392fa1f1 100644
--- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOWriteTest.java
+++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOWriteTest.java
@@ -20,8 +20,11 @@
import static org.apache.beam.sdk.values.TypeDescriptors.strings;
import com.solacesystems.jcsmp.DeliveryMode;
+import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
@@ -82,6 +85,31 @@ private PCollection getRecords(Pipeline p) {
.via(kv -> SolaceDataUtils.getSolaceRecord(kv.getValue(), kv.getKey())));
}
+ private PCollection getRecordsForEachPayloadTypes(Pipeline p) {
+ TestStream.Builder kvBuilder =
+ TestStream.create(AvroCoder.of(Record.PayloadType.class)).advanceWatermarkTo(Instant.EPOCH);
+
+ for (var payloadType : Record.PayloadType.values()) {
+ kvBuilder =
+ kvBuilder.addElements(payloadType).advanceProcessingTime(Duration.standardSeconds(60));
+ }
+
+ TestStream testStream = kvBuilder.advanceWatermarkToInfinity();
+
+ return p.apply("Test stream ", testStream)
+ .apply(
+ "To Record",
+ MapElements.into(TypeDescriptor.of(Record.class))
+ .via(
+ payloadType ->
+ Solace.Record.builder()
+ .setMessageId(payloadType.name().toLowerCase())
+ .setPayloadType(payloadType)
+ .setPayload(
+ ("payload-" + payloadType.name()).getBytes(StandardCharsets.UTF_8))
+ .build()));
+ }
+
private SolaceOutput getWriteTransform(
SubmissionMode mode,
WriterType writerType,
@@ -172,6 +200,61 @@ public void testWriteThroughputBatched() throws Exception {
pipeline.run();
}
+ @Test
+ public void testWriteMixedPayloadTypesStreaming() throws Exception {
+ PCollection records = getRecordsForEachPayloadTypes(pipeline);
+
+ ErrorHandler> errorHandler =
+ pipeline.registerBadRecordErrorHandler(new ErrorSinkTransform());
+
+ SolaceOutput output =
+ records.apply(
+ "Write mixed records",
+ SolaceIO.write()
+ .to(Solace.Queue.fromName("queue"))
+ .withSubmissionMode(SubmissionMode.LOWER_LATENCY)
+ .withWriterType(WriterType.STREAMING)
+ .withDeliveryMode(DeliveryMode.PERSISTENT)
+ .withSessionServiceFactory(MockSessionServiceFactory.builder().build())
+ .withErrorHandler(errorHandler));
+
+ var expectedIds =
+ Stream.of(Record.PayloadType.values())
+ .map(payloadType -> payloadType.name().toLowerCase())
+ .collect(Collectors.toList());
+ PAssert.that(getIdsPCollection(output)).containsInAnyOrder(expectedIds);
+ errorHandler.close();
+ PAssert.that(errorHandler.getOutput()).empty();
+ pipeline.run();
+ }
+
+ @Test
+ public void testWriteMixedPayloadTypesBatched() throws Exception {
+ PCollection records = getRecordsForEachPayloadTypes(pipeline);
+
+ ErrorHandler> errorHandler =
+ pipeline.registerBadRecordErrorHandler(new ErrorSinkTransform());
+ SolaceOutput output =
+ records.apply(
+ "Write mixed records",
+ SolaceIO.write()
+ .to(Solace.Queue.fromName("queue"))
+ .withSubmissionMode(SubmissionMode.HIGHER_THROUGHPUT)
+ .withWriterType(WriterType.BATCHED)
+ .withDeliveryMode(DeliveryMode.PERSISTENT)
+ .withSessionServiceFactory(MockSessionServiceFactory.builder().build())
+ .withErrorHandler(errorHandler));
+
+ var expectedIds =
+ Stream.of(Record.PayloadType.values())
+ .map(payloadType -> payloadType.name().toLowerCase())
+ .collect(Collectors.toList());
+ PAssert.that(getIdsPCollection(output)).containsInAnyOrder(expectedIds);
+ errorHandler.close();
+ PAssert.that(errorHandler.getOutput()).empty();
+ pipeline.run();
+ }
+
@Test
public void testWriteWithFailedRecords() throws Exception {
SubmissionMode mode = SubmissionMode.HIGHER_THROUGHPUT;
diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java
new file mode 100644
index 000000000000..cc6567b4c88c
--- /dev/null
+++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java
@@ -0,0 +1,313 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.solace.data;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import com.solacesystems.jcsmp.BytesMessage;
+import com.solacesystems.jcsmp.BytesXMLMessage;
+import com.solacesystems.jcsmp.DeliveryMode;
+import com.solacesystems.jcsmp.JCSMPFactory;
+import com.solacesystems.jcsmp.TextMessage;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import org.apache.beam.sdk.io.solace.broker.MessageProducerUtils;
+import org.apache.beam.sdk.io.solace.data.Solace.Record;
+import org.apache.beam.sdk.io.solace.data.Solace.Record.PayloadType;
+import org.junit.Test;
+
+public class SolaceRecordMapperTest {
+
+ @Test
+ public void testNullMessage() {
+ assertNull(Solace.SolaceRecordMapper.toRecord(null));
+ }
+
+ @Test
+ public void testTextMessage() {
+ TextMessage message = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);
+ message.setApplicationMessageId("id");
+ message.setText("héllo");
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertEquals(Record.PayloadType.TEXT, record.getPayloadType());
+ assertArrayEquals("héllo".getBytes(StandardCharsets.UTF_8), record.getPayload());
+ assertArrayEquals(new byte[0], record.getAttachmentBytes());
+ }
+
+ @Test
+ public void testBytesMessage() {
+ byte[] payload = new byte[] {0, 1, (byte) 255};
+ BytesMessage message = JCSMPFactory.onlyInstance().createMessage(BytesMessage.class);
+ message.setApplicationMessageId("id");
+ message.setData(payload);
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertEquals(Record.PayloadType.BYTES, record.getPayloadType());
+ assertArrayEquals(payload, record.getPayload());
+ assertArrayEquals(new byte[0], record.getAttachmentBytes());
+ }
+
+ @Test
+ public void testBytesXmlMessageAndAttachment() {
+ Record source =
+ Solace.Record.builder()
+ .setMessageId("id")
+ .setPayload(new byte[] {1, 2})
+ .setAttachmentBytes(new byte[] {3, 4})
+ .build();
+ BytesXMLMessage message =
+ MessageProducerUtils.createMessage(source, false, DeliveryMode.DIRECT);
+ message.setReadOnly();
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertEquals(Record.PayloadType.BYTES_XML, record.getPayloadType());
+ assertArrayEquals(new byte[] {1, 2}, Arrays.copyOf(record.getPayload(), 2));
+ assertArrayEquals(new byte[] {3, 4}, record.getAttachmentBytes());
+ }
+
+ @Test
+ public void testNullTextPayload() {
+ TextMessage message = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);
+ message.setApplicationMessageId("id");
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertEquals(Record.PayloadType.TEXT, record.getPayloadType());
+ assertArrayEquals(new byte[0], record.getPayload());
+ }
+
+ @Test
+ public void testNullBytesPayload() {
+ BytesMessage message = JCSMPFactory.onlyInstance().createMessage(BytesMessage.class);
+ message.setApplicationMessageId("id");
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertEquals(Record.PayloadType.BYTES, record.getPayloadType());
+ assertArrayEquals(new byte[0], record.getPayload());
+ }
+
+ @Test
+ public void testEmptyBytesXmlMessage() {
+ BytesXMLMessage message = JCSMPFactory.onlyInstance().createBytesXMLMessage();
+ message.setApplicationMessageId("id");
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertEquals(Record.PayloadType.BYTES_XML, record.getPayloadType());
+ assertArrayEquals(new byte[0], record.getPayload());
+ assertArrayEquals(new byte[0], record.getAttachmentBytes());
+ }
+
+ @Test
+ public void testMapMessageMetadata() {
+ TextMessage message = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);
+ message.setApplicationMessageId("id");
+ message.setText("hello");
+ message.setExpiration(123L);
+ message.setPriority(7);
+ message.setReplyTo(JCSMPFactory.onlyInstance().createQueue("reply-queue"));
+ message.setSenderTimestamp(456L);
+ message.setTimeToLive(789L);
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertEquals("id", record.getMessageId());
+ assertEquals(123L, record.getExpiration());
+ assertEquals(7, record.getPriority());
+ assertEquals(false, record.getRedelivered());
+ assertEquals("reply-queue", record.getReplyTo().getName());
+ assertEquals(Solace.DestinationType.QUEUE, record.getReplyTo().getType());
+ assertEquals(Long.valueOf(456L), record.getSenderTimestamp());
+ assertEquals(789L, record.getTimeToLive());
+ }
+
+ @Test
+ public void testMapTextRecord() {
+ Record record =
+ Record.builder().setMessageId("id").setText("héllo").setSenderTimestamp(1L).build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertTrue(msg instanceof TextMessage);
+ assertEquals("héllo", ((TextMessage) msg).getText());
+ }
+
+ @Test
+ public void testMapTextRecordWithEmptyText() {
+ Record record = Record.builder().setMessageId("id").setText("").setSenderTimestamp(1L).build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertTrue(msg instanceof TextMessage);
+ assertEquals("", ((TextMessage) msg).getText());
+ }
+
+ @Test
+ public void testMapBytesRecord() {
+ byte[] payload = new byte[] {0, 1, (byte) 255};
+ Record record =
+ Record.builder()
+ .setMessageId("id")
+ .setPayload(payload)
+ .setPayloadType(PayloadType.BYTES)
+ .setSenderTimestamp(1L)
+ .build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertTrue(msg instanceof BytesMessage);
+ assertArrayEquals(payload, ((BytesMessage) msg).getData());
+ }
+
+ @Test
+ public void testMapBytesXmlRecord() {
+ byte[] payload = new byte[] {1, 2, 3};
+ Record record =
+ Record.builder().setMessageId("id").setPayload(payload).setSenderTimestamp(1L).build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertArrayEquals(payload, Arrays.copyOf(msg.getBytes(), msg.getContentLength()));
+ }
+
+ @Test
+ public void testMapBytesXmlRecordWithAttachment() {
+ byte[] payload = new byte[] {1, 2};
+ byte[] attachment = new byte[] {3, 4};
+ Record record =
+ Record.builder()
+ .setMessageId("id")
+ .setPayload(payload)
+ .setAttachmentBytes(attachment)
+ .setSenderTimestamp(1L)
+ .build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertEquals(attachment.length, msg.getAttachmentContentLength());
+ assertArrayEquals(attachment, msg.getAttachmentByteBuffer().array());
+ }
+
+ @Test
+ public void testMapBytesXmlRecordWithEmptyAttachment() {
+ Record record =
+ Record.builder()
+ .setMessageId("id")
+ .setPayload(new byte[] {1})
+ .setSenderTimestamp(1L)
+ .build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertEquals(0, msg.getAttachmentContentLength());
+ }
+
+ @Test
+ public void testMapRecordMetadata() {
+ Record record =
+ Record.builder().setMessageId("id").setText("hello").setSenderTimestamp(1L).build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertEquals("id", msg.getApplicationMessageId());
+ assertEquals(Long.valueOf(1L), Long.valueOf(msg.getSenderTimestamp()));
+ }
+
+ @Test
+ public void testToMessageDefaultsSenderTimestamp() {
+ Record record = Record.builder().setMessageId("id").setText("hello").build();
+
+ long before = System.currentTimeMillis();
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+ long after = System.currentTimeMillis();
+
+ assertTrue(msg.getSenderTimestamp() >= before && msg.getSenderTimestamp() <= after);
+ }
+
+ @Test
+ public void testToMessageDoesNotSetPublishingFields() {
+ Record record =
+ Record.builder().setMessageId("id").setText("hello").setSenderTimestamp(1L).build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertNull(msg.getCorrelationKey());
+ }
+
+ // ---------------------------------------------------------------------------
+ // round-trip
+ // ---------------------------------------------------------------------------
+ @Test
+ public void testRoundTripTextPayload() {
+ Record original =
+ Record.builder().setMessageId("id").setText("héllo").setSenderTimestamp(1L).build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(original);
+ msg.setApplicationMessageId("id");
+ Record decoded = Solace.SolaceRecordMapper.toRecord(msg);
+
+ assertEquals(original.getPayloadType(), decoded.getPayloadType());
+ assertArrayEquals(original.getPayload(), decoded.getPayload());
+ }
+
+ @Test
+ public void testRoundTripBytesPayload() {
+ byte[] payload = new byte[] {10, 20, 30};
+ Record original =
+ Record.builder()
+ .setMessageId("id")
+ .setPayload(payload)
+ .setPayloadType(PayloadType.BYTES)
+ .setSenderTimestamp(1L)
+ .build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(original);
+ msg.setApplicationMessageId("id");
+ Record decoded = Solace.SolaceRecordMapper.toRecord(msg);
+
+ assertEquals(PayloadType.BYTES, decoded.getPayloadType());
+ assertArrayEquals(payload, decoded.getPayload());
+ }
+
+ @Test
+ public void testRoundTripBytesXmlPayloadWithAttachment() {
+ Record original =
+ Record.builder()
+ .setMessageId("id")
+ .setPayload(new byte[] {1, 2})
+ .setAttachmentBytes(new byte[] {3, 4})
+ .setSenderTimestamp(1L)
+ .build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(original);
+ msg.setApplicationMessageId("id");
+ Record decoded = Solace.SolaceRecordMapper.toRecord(msg);
+
+ assertEquals(PayloadType.BYTES_XML, decoded.getPayloadType());
+ assertArrayEquals(new byte[] {1, 2}, Arrays.copyOf(decoded.getPayload(), 2));
+ assertArrayEquals(new byte[] {3, 4}, decoded.getAttachmentBytes());
+ }
+}
diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordTest.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordTest.java
new file mode 100644
index 000000000000..5b521f7d4e68
--- /dev/null
+++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.solace.data;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+import java.nio.charset.StandardCharsets;
+import org.apache.beam.sdk.io.solace.data.Solace.Record;
+import org.junit.Test;
+
+public class SolaceRecordTest {
+
+ @Test
+ public void testDefaultPayloadType() {
+ Record record = Record.builder().setMessageId("id").setPayload(new byte[0]).build();
+
+ assertEquals(Record.PayloadType.BYTES_XML, record.getPayloadType());
+ }
+
+ @Test
+ public void testSetTextPayload() {
+ Record record = Record.builder().setMessageId("id").setText("héllo").build();
+
+ assertEquals(Record.PayloadType.TEXT, record.getPayloadType());
+ assertArrayEquals("héllo".getBytes(StandardCharsets.UTF_8), record.getPayload());
+ assertEquals("héllo", record.getText());
+ }
+}