SolaceIO: support binary content (text, bytes) data payload

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
 

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java

@@ -84,7 +84,7 @@
  * <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 @@
  * 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=
@@ -458,7 +465,7 @@ public static Read<Solace.Record> read() {
     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));

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 static BytesXMLMessage createBytesXMLMessage(
       // Use only a string as correlation key
       msg.setCorrelationKey(record.getMessageId());
     }
-    msg.setApplicationMessageId(record.getMessageId());
     return msg;
   }
 
@@ -100,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;
     }

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<Integer> publish =
         () -> {
           producer.send(msg, topicOrQueue);

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