ngibanel commented on code in PR #40108:
URL: https://github.com/apache/beam/pull/40108#discussion_r4004885413
##########
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);
+ } catch (SDTException e) {
+ LOG.error("Could not read user property '{}'.", key, e);
Review Comment:
ok, that means the consumer must enable the dead message queue in Solace
(which is a best practice) to avoid losing messages if metadata cannot be
deserialized. Otherwise the broker will redeliver the message until the max
retry count is reached and then message will be discarded and lost.
--
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]
