ngibanel commented on code in PR #40108:
URL: https://github.com/apache/beam/pull/40108#discussion_r4026143137
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java:
##########
@@ -599,5 +623,47 @@ private static byte[] readAttachment(BytesXMLMessage msg) {
buffer.get(attachment);
return attachment;
}
+
+ private static Map<String, String> getUserProperties(@Nullable SDTMap
properties) {
+ if (properties == null || properties.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Map<String, String> userProperties = new HashMap<>();
+ for (String key : properties.keySet()) {
+ String value = stringifyUserProperty(properties, key);
+ if (value == null) {
+ LOG.warn("User property '{}' has a null value, skipping.", key);
+ continue;
+ }
+ userProperties.put(key, value);
+ }
+ return Collections.unmodifiableMap(userProperties);
+ }
+
+ private static @Nullable String stringifyUserProperty(SDTMap properties,
String key) {
+ try {
+ Object value = properties.get(key);
+ if (value == null) {
+ return null;
+ }
+ return String.valueOf(value);
Review Comment:
Yes, the issue is that Beam cannot infer the schema because Map and Stream
are recursively supported in JCSMP user properties, which creates schema
inference challenges.
To keep a clean public API that exposes user properties as a Map<String,
UserPropertyValue>, one possible workaround would be to store the properties
internally in a schema-safe representation where Beam can infer the schema. For
example, user properties could be encoded as a list of key/value pairs, with
the value represented as the serialized byte form of the underlying JCSMP
property value.
The Beam schema would then rely on this internal representation, while the
SDK would expose a more convenient typed API:
```java
/** Gets the schema-safe transport representation of the user properties. */
@SchemaFieldNumber("14")
abstract List<UserProperty> getUserProperties();
/**
* Gets the typed, SDK-independent user properties of the message.
*
* <p>This accessor is excluded from the Beam schema because {@link
UserPropertyValue}
* recursively represents maps and streams.
*
* @return The user properties, or an empty map if the message carries
none.
*/
@SchemaIgnore
public final Map<String, UserPropertyValue> getUserPropertiesMap() {
return
SolaceUserPropertiesMapper.toUserPropertyValueMap(getUserProperties());
}
```
Do you see a better approach for handling JCSMP user properties that may
contain nested maps or streams?
Another option would be to explicitly not support map and stream property
values. To be honest, I'm not even sure how Solace maps a JCSMP Stream when
bridging to other protocols such as AMQP or MQTT 5.0, nor whether such property
types provide much practical interoperability value. If they are
protocol-specific constructs, excluding them may be a reasonable simplification
for the SDK.
--
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]
