This is an automated email from the ASF dual-hosted git repository.
stankiewicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new e6d7f23de91 SolaceIO: support binary content (text, bytes) data
payload (#39876)
e6d7f23de91 is described below
commit e6d7f23de9106d62184779524184351df75d3f1b
Author: Nicolas Gibanel <[email protected]>
AuthorDate: Tue Sep 1 08:33:57 2026 +0200
SolaceIO: support binary content (text, bytes) data payload (#39876)
* SolaceIO: support binary content (text, bytes) data payload
* fixup: rename createBytesXMLMessage() to createMessage()
* fixup: simplify and optimize read bytes
---------
Co-authored-by: Nicolas Gibanel <[email protected]>
---
CHANGES.md | 1 +
.../org/apache/beam/sdk/io/solace/SolaceIO.java | 11 +-
.../sdk/io/solace/broker/MessageProducerUtils.java | 16 +-
.../io/solace/broker/SolaceMessageProducer.java | 4 +-
.../org/apache/beam/sdk/io/solace/data/Solace.java | 189 +++++++++++--
.../beam/sdk/io/solace/SolaceIOWriteTest.java | 83 ++++++
.../sdk/io/solace/data/SolaceRecordMapperTest.java | 313 +++++++++++++++++++++
.../beam/sdk/io/solace/data/SolaceRecordTest.java | 44 +++
8 files changed, 618 insertions(+), 43 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 6357802c0ee..5b52e9319c8 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -67,6 +67,7 @@
* Support for X source added (Java/Python)
([#X](https://github.com/apache/beam/issues/X)).
* Added `schema_update_options` to `WriteToBigQuery` file loads, allowing
BigQuery load jobs to add nullable fields or relax required fields when
appending data (Python) ([#21141](https://github.com/apache/beam/issues/21141)).
* BigQueryIO now supports reading BigQuery Lakehouse runtime catalog (BigLake
metastore) Iceberg tables with the Storage Read API, using 4-part
`project.catalog.namespace.table` identifiers (or a `TableReference` with a
composite `catalog.namespace` dataset id). Previously such references were
silently mis-parsed (Java)
([#39597](https://github.com/apache/beam/issues/39597)) .
+* SolaceIO now supports reading and writing binary and text content data
payload (Java) ([#39875](https://github.com/apache/beam/issues/39875)).
* ClickHouseIO: support writing `Decimal(P, S)` / `Decimal32/64/128/256`
columns (Java) ([#39840](https://github.com/apache/beam/issues/39840)).
## New Features / Improvements
diff --git
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
index c0ecac96fdb..57f65ae6265 100644
---
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
+++
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
@@ -84,7 +84,7 @@ import org.slf4j.LoggerFactory;
* <h3>No-argument {@link SolaceIO#read()} top-level method</h3>
*
* <p>This method returns a PCollection of {@link Solace.Record} objects. It
uses a default mapper
- * ({@link SolaceRecordMapper#map(BytesXMLMessage)}) to map from the received
{@link
+ * ({@link SolaceRecordMapper#toRecord(BytesXMLMessage)}) to map from the
received {@link
* BytesXMLMessage} from Solace, to the {@link Solace.Record} objects.
*
* <p>By default, it also uses a {@link BytesXMLMessage#getSenderTimestamp()}
for watermark
@@ -221,6 +221,13 @@ import org.slf4j.LoggerFactory;
* also use {@link #write(SerializableFunction)} to specify a format function
to convert the input
* type to {@link Solace.Record}.
*
+ * <p>Each record can select its JCSMP payload representation through {@link
+ * Solace.Record.PayloadType}. The default is {@link
Solace.Record.PayloadType#BYTES_XML}, which
+ * preserves the historical behavior of writing the byte array with {@code
+ * BytesXMLMessage.writeBytes}. Use {@code setText(String)} to create a UTF-8
{@link
+ * com.solacesystems.jcsmp.TextMessage}, or select {@link
Solace.Record.PayloadType#BYTES} to
+ * publish the byte array with a JCSMP {@link
com.solacesystems.jcsmp.BytesMessage}.
+ *
* <h3>Writing to a static topic or queue</h3>
*
* <p>The connector uses the <a href=
@@ -460,7 +467,7 @@ public class SolaceIO {
return new Read<Solace.Record>(
Read.Configuration.<Solace.Record>builder()
.setTypeDescriptor(TypeDescriptor.of(Solace.Record.class))
- .setParseFn(SolaceRecordMapper::map)
+ .setParseFn(SolaceRecordMapper::toRecord)
.setTimestampFn(SENDER_TIMESTAMP_FUNCTION)
.setDeduplicateRecords(DEFAULT_DEDUPLICATE_RECORDS)
.setWatermarkIdleDurationThreshold(DEFAULT_WATERMARK_IDLE_DURATION_THRESHOLD)
diff --git
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/MessageProducerUtils.java
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/MessageProducerUtils.java
index dd4610910ff..1b258dfeb62 100644
---
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/MessageProducerUtils.java
+++
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/MessageProducerUtils.java
@@ -40,18 +40,9 @@ public class MessageProducerUtils {
* @param deliveryMode The {@link DeliveryMode} used to publish the message.
* @return A {@link BytesXMLMessage} that can be sent to Solace "as is".
*/
- public static BytesXMLMessage createBytesXMLMessage(
+ public static BytesXMLMessage createMessage(
Solace.Record record, boolean useCorrelationKeyLatency, DeliveryMode
deliveryMode) {
- JCSMPFactory jcsmpFactory = JCSMPFactory.onlyInstance();
- BytesXMLMessage msg = jcsmpFactory.createBytesXMLMessage();
- byte[] payload = record.getPayload();
- msg.writeBytes(payload);
-
- Long senderTimestamp = record.getSenderTimestamp();
- if (senderTimestamp == null) {
- senderTimestamp = System.currentTimeMillis();
- }
- msg.setSenderTimestamp(senderTimestamp);
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
msg.setDeliveryMode(deliveryMode);
if (useCorrelationKeyLatency) {
Solace.CorrelationKey key =
@@ -64,7 +55,6 @@ public class MessageProducerUtils {
// Use only a string as correlation key
msg.setCorrelationKey(record.getMessageId());
}
- msg.setApplicationMessageId(record.getMessageId());
return msg;
}
@@ -100,7 +90,7 @@ public class MessageProducerUtils {
JCSMPSendMultipleEntry entry =
JCSMPFactory.onlyInstance()
.createSendMultipleEntry(
- createBytesXMLMessage(record, useCorrelationKeyLatency,
deliveryMode),
+ createMessage(record, useCorrelationKeyLatency,
deliveryMode),
destinationFn.apply(record));
entries[i] = entry;
}
diff --git
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceMessageProducer.java
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceMessageProducer.java
index b3806b5afae..227f72fb2b6 100644
---
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceMessageProducer.java
+++
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceMessageProducer.java
@@ -17,8 +17,8 @@
*/
package org.apache.beam.sdk.io.solace.broker;
-import static
org.apache.beam.sdk.io.solace.broker.MessageProducerUtils.createBytesXMLMessage;
import static
org.apache.beam.sdk.io.solace.broker.MessageProducerUtils.createJCSMPSendMultipleEntry;
+import static
org.apache.beam.sdk.io.solace.broker.MessageProducerUtils.createMessage;
import com.solacesystems.jcsmp.BytesXMLMessage;
import com.solacesystems.jcsmp.DeliveryMode;
@@ -50,7 +50,7 @@ public class SolaceMessageProducer implements MessageProducer
{
Destination topicOrQueue,
boolean useCorrelationKeyLatency,
DeliveryMode deliveryMode) {
- BytesXMLMessage msg = createBytesXMLMessage(record,
useCorrelationKeyLatency, deliveryMode);
+ BytesXMLMessage msg = createMessage(record, useCorrelationKeyLatency,
deliveryMode);
Callable<Integer> 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 e6cd35b63b4..15fe06103fb 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 class Solace {
@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 class Solace {
@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 class Solace {
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 class Solace {
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 class Solace {
*/
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 class Solace {
* @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 class Solace {
msg.getReplicationGroupMessageId() != null
? msg.getReplicationGroupMessageId().toString()
: null)
- .setAttachmentBytes(attachmentBytesStream.toByteArray())
.build();
}
@@ -462,5 +496,108 @@ public class Solace {
}
return destinationBuilder.build();
}
+
+ /**
+ * Maps a {@link Record} to a {@link BytesXMLMessage}.
+ *
+ * <p>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 e92657c3c3d..3cdc392fa1f 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 @@ package org.apache.beam.sdk.io.solace;
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 @@ public class SolaceIOWriteTest {
.via(kv -> SolaceDataUtils.getSolaceRecord(kv.getValue(),
kv.getKey())));
}
+ private PCollection<Record> getRecordsForEachPayloadTypes(Pipeline p) {
+ TestStream.Builder<Record.PayloadType> 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<Record.PayloadType> 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 class SolaceIOWriteTest {
pipeline.run();
}
+ @Test
+ public void testWriteMixedPayloadTypesStreaming() throws Exception {
+ PCollection<Record> records = getRecordsForEachPayloadTypes(pipeline);
+
+ ErrorHandler<BadRecord, PCollection<Long>> 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<Record> records = getRecordsForEachPayloadTypes(pipeline);
+
+ ErrorHandler<BadRecord, PCollection<Long>> 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 00000000000..cc6567b4c88
--- /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 00000000000..5b521f7d4e6
--- /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());
+ }
+}
