From 398149eed2e30db0278bee3d2c1d136682b79c43 Mon Sep 17 00:00:00 2001 From: Nicolas Gibanel Date: Mon, 24 Aug 2026 12:00:56 +0200 Subject: [PATCH 1/3] SolaceIO: support binary content (text, bytes) data payload --- CHANGES.md | 1 + .../apache/beam/sdk/io/solace/SolaceIO.java | 11 +- .../solace/broker/MessageProducerUtils.java | 12 +- .../beam/sdk/io/solace/data/Solace.java | 196 +++++++++-- .../beam/sdk/io/solace/SolaceIOWriteTest.java | 83 +++++ .../solace/data/SolaceRecordMapperTest.java | 313 ++++++++++++++++++ .../sdk/io/solace/data/SolaceRecordTest.java | 44 +++ 7 files changed, 623 insertions(+), 37 deletions(-) create mode 100644 sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java create mode 100644 sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordTest.java diff --git a/CHANGES.md b/CHANGES.md index 3994aea4d938..2346e1505c8d 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -66,6 +66,7 @@ * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). * 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)). ## 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 63509126022b..9b483a4cd131 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 @@ *

No-argument {@link SolaceIO#read()} top-level method

* *

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. * *

By default, it also uses a {@link BytesXMLMessage#getSenderTimestamp()} for watermark @@ -221,6 +221,13 @@ * also use {@link #write(SerializableFunction)} to specify a format function to convert the input * type to {@link Solace.Record}. * + *

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}. + * *

Writing to a static topic or queue

* *

The connector uses the read() { return new Read( Read.Configuration.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 dd4610910ff4..6f04b131807a 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 @@ -42,16 +42,7 @@ public class MessageProducerUtils { */ public static BytesXMLMessage createBytesXMLMessage( 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 static BytesXMLMessage createBytesXMLMessage( // Use only a string as correlation key msg.setCorrelationKey(record.getMessageId()); } - msg.setApplicationMessageId(record.getMessageId()); return msg; } 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..2d85eb553ce6 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,20 @@ 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 com.solacesystems.jcsmp.JCSMPFactory; +import com.solacesystems.jcsmp.TextMessage; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; 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 +120,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 +273,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 +302,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 +335,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 +440,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 +450,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 +474,6 @@ public static class SolaceRecordMapper { msg.getReplicationGroupMessageId() != null ? msg.getReplicationGroupMessageId().toString() : null) - .setAttachmentBytes(attachmentBytesStream.toByteArray()) .build(); } @@ -462,5 +497,118 @@ 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]; + } + try { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + out.write(msg.getBytes()); + return out.toByteArray(); + } catch (IOException e) { + LOG.error("Could not read bytes from BytesXMLMessage.", e); + return new byte[0]; + } + } + + private static byte[] readAttachment(BytesXMLMessage msg) { + if (msg.getAttachmentContentLength() == 0) { + return new byte[0]; + } + try { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + out.write(msg.getAttachmentByteBuffer().array()); + return out.toByteArray(); + } catch (IOException e) { + LOG.error("Could not read attachment from BytesXMLMessage.", e); + return new byte[0]; + } + } } } 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..b5c281593c62 --- /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.createBytesXMLMessage(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()); + } +} From 406195187b682e802a6e48b5784b6083741baa33 Mon Sep 17 00:00:00 2001 From: Nicolas Gibanel Date: Fri, 28 Aug 2026 16:28:27 +0200 Subject: [PATCH 2/3] fixup: rename createBytesXMLMessage() to createMessage() --- .../beam/sdk/io/solace/broker/MessageProducerUtils.java | 4 ++-- .../beam/sdk/io/solace/broker/SolaceMessageProducer.java | 4 ++-- .../beam/sdk/io/solace/data/SolaceRecordMapperTest.java | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) 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 6f04b131807a..1b258dfeb620 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,7 +40,7 @@ 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) { BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record); msg.setDeliveryMode(deliveryMode); @@ -90,7 +90,7 @@ public static JCSMPSendMultipleEntry[] createJCSMPSendMultipleEntry( 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 b3806b5afae9..227f72fb2b61 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 void publishSingleMessage( Destination topicOrQueue, boolean useCorrelationKeyLatency, DeliveryMode deliveryMode) { - BytesXMLMessage msg = createBytesXMLMessage(record, useCorrelationKeyLatency, deliveryMode); + BytesXMLMessage msg = createMessage(record, useCorrelationKeyLatency, deliveryMode); Callable publish = () -> { producer.send(msg, topicOrQueue); 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 index b5c281593c62..cc6567b4c88c 100644 --- 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 @@ -77,7 +77,7 @@ public void testBytesXmlMessageAndAttachment() { .setAttachmentBytes(new byte[] {3, 4}) .build(); BytesXMLMessage message = - MessageProducerUtils.createBytesXMLMessage(source, false, DeliveryMode.DIRECT); + MessageProducerUtils.createMessage(source, false, DeliveryMode.DIRECT); message.setReadOnly(); Record record = Solace.SolaceRecordMapper.toRecord(message); From 1ed3b6ea3befab92a23e2be49c36829934174b34 Mon Sep 17 00:00:00 2001 From: Nicolas Gibanel Date: Fri, 28 Aug 2026 16:33:57 +0200 Subject: [PATCH 3/3] fixup: simplify and optimize read bytes --- .../beam/sdk/io/solace/data/Solace.java | 25 ++++++------------- 1 file changed, 7 insertions(+), 18 deletions(-) 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 2d85eb553ce6..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 @@ -22,12 +22,11 @@ import com.solacesystems.jcsmp.BytesXMLMessage; import com.solacesystems.jcsmp.JCSMPFactory; import com.solacesystems.jcsmp.TextMessage; -import java.io.ByteArrayOutputStream; -import java.io.IOException; 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; @@ -587,28 +586,18 @@ private static byte[] readBytes(BytesXMLMessage msg) { if (msg.getContentLength() == 0) { return new byte[0]; } - try { - ByteArrayOutputStream out = new ByteArrayOutputStream(); - out.write(msg.getBytes()); - return out.toByteArray(); - } catch (IOException e) { - LOG.error("Could not read bytes from BytesXMLMessage.", e); - 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]; } - try { - ByteArrayOutputStream out = new ByteArrayOutputStream(); - out.write(msg.getAttachmentByteBuffer().array()); - return out.toByteArray(); - } catch (IOException e) { - LOG.error("Could not read attachment from BytesXMLMessage.", e); - return new byte[0]; - } + + ByteBuffer buffer = msg.getAttachmentByteBuffer(); + byte[] attachment = new byte[buffer.remaining()]; + buffer.get(attachment); + return attachment; } } }