ngibanel opened a new pull request, #39876: URL: https://github.com/apache/beam/pull/39876
## Summary Extend Solace.Record with a per-record payload type to support JCSMP TextMessage and BytesMessage type for reading and writing data. The default remains through BytesXMLMessage to preserve existing users' behavior and source compatibility. Fixes #39875 ## Motivation Solace messages support several distinct payload types (https://docs.solace.com/API/API-Developer-Guide/Adding-Data-Payloads.htm): - XML content - data added to the XML part of the message, processed and inspectable by the broker. - Binary attachment - data added as a binary attachment; the broker passes it through as-is without processing, transformation, or involvement in subscription matching. This is the recommended approach for arbitrary application payloads. - Text - a UTF-8 string payload carried by a dedicated TextMessage type. - User data / User property map - small header-level fields for metadata (out of scope of this PR) The previous SolaceIO implementation always wrote the payload using BytesXMLMessage.writeBytes(), which places data in the XML content part of the message. This is a legacy behavior: it works, but it is not the recommended way to carry arbitrary binary payloads, and it creates interoperability issues. When a Solace message is produced by other SDKs - JMS in particular - the payload is serialized into the binary attachment, not the XML content part. A consumer using SolaceIO would therefore receive an empty payload when reading messages written by a JMS producer, and vice versa. This PR aligns SolaceIO with the recommended Solace practices by introducing explicit payload type selection, covering: | Payload type | JCSMP type | Solace payload slot | |---------------------|-------------------|--------------------------------------------| | `BYTES_XML` (default) | `BytesXMLMessage` | XML content (legacy, backward-compatible) | | `TEXT` | `TextMessage` | Text payload | | `BYTES` | `BytesMessage` | Binary attachment | ## Changes ### 1. New `PayloadType` field on `Solace.Record` A `Record.PayloadType` enum is introduced with three values: - `BYTES_XML` (default) - legacy behavior, fully backward-compatible. - `TEXT` - UTF-8 text payload. - `BYTES` - raw binary payload (binary attachment). The payload is always stored as `byte[]` inside `Record`, regardless of type. For `TEXT`, those bytes are the UTF-8 encoding of the string. This avoids adding a separate `String` field to the Beam schema and keeps a uniform data model. The `getText()` accessor decodes on demand with strict UTF-8 validation. ### 2. Builder convenience method `setText(String)` A shorthand on `Record.Builder` that encodes the string to UTF-8, stores the bytes in `payload`, and sets `payloadType` to `TEXT` in a single call. ### 3. `SolaceRecordMapper` refactoring - `map()` renamed to `toRecord()` (more expressive, symmetric with the new `toMessage()`). - New public method `toMessage(Record)`: centralizes the `Record → BytesXMLMessage` conversion previously scattered in `MessageProducerUtils`. It maps only the fields that are common to both the `Record` model and a JCSMP message: the payload (dispatched by `PayloadType`), `senderTimestamp` (defaulting to `System.currentTimeMillis()`), and `applicationMessageId`. Protocol-specific publishing fields - delivery mode, correlation key, etc. - are intentionally left out and remain the responsibility of `MessageProducerUtils`. - Private `decodePayload(BytesXMLMessage)`: detects the actual JCSMP type via `instanceof` (`TextMessage`, `BytesMessage`, or fallback to `BytesXMLMessage`) and builds the appropriate `Record.Builder`. - Private `encodePayload(Record)`: creates the correct JCSMP object based on `PayloadType`. ### 4. `MessageProducerUtils` simplification Message construction logic is delegated to `SolaceRecordMapper.toMessage()`. Only protocol-specific publishing fields are set. ## Backward compatibility - `Record.builder()` defaults to `PayloadType.BYTES_XML` ; existing pipelines require no changes. - `SolaceRecordMapper.map()` is renamed to `toRecord()`; callers using the default `read()` API are unaffected (the internal reference is updated). ------------------------ Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily: - [x] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead. - [x] Update `CHANGES.md` with noteworthy changes. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf). See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier). To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
