diff --git a/CHANGES.md b/CHANGES.md index 47da5f81fb02..cd4ea53eb76c 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -99,6 +99,7 @@ * 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)). * ClickHouseIO: support writing `Decimal(P, S)` / `Decimal32/64/128/256` columns (Java) ([#39840](https://github.com/apache/beam/issues/39840)). +* SolaceIO now supports reading and writing user properties (message metadata) (Java) ([#40099](https://github.com/apache/beam/issues/40099)). * [IcebergIO] AddFiles (`IcebergAddFiles` in YAML) can evolve the table schema before registering files, with `schema_evolution_options`, `required_columns`, `incompatible_schema_handling` and `unverifiable_file_handling` (Java/YAML, batch only) ([#40144](https://github.com/apache/beam/issues/40144)). ## New Features / Improvements diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceUserPropertiesMapper.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceUserPropertiesMapper.java new file mode 100644 index 000000000000..45e5f60e9859 --- /dev/null +++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceUserPropertiesMapper.java @@ -0,0 +1,200 @@ +/* + * 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.broker; + +import com.solacesystems.common.util.ByteArray; +import com.solacesystems.jcsmp.Destination; +import com.solacesystems.jcsmp.JCSMPFactory; +import com.solacesystems.jcsmp.SDTException; +import com.solacesystems.jcsmp.SDTMap; +import com.solacesystems.jcsmp.Topic; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.apache.beam.sdk.io.solace.data.Solace; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Bytes; +import org.checkerframework.checker.nullness.qual.NonNull; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public final class SolaceUserPropertiesMapper { + + private static final Logger LOG = LoggerFactory.getLogger(SolaceUserPropertiesMapper.class); + + private SolaceUserPropertiesMapper() {} + + public static Map toUserPropertyValueMap( + @Nullable SDTMap properties) { + if (properties == null || properties.isEmpty()) { + return Collections.emptyMap(); + } + + Map userProperties = new HashMap<>(); + for (String key : properties.keySet()) { + try { + Object value = properties.get(key); + if (value == null) { + continue; + } + + Solace.UserPropertyValue userPropertyValue = toUserPropertyValue(value); + if (userPropertyValue.getKind() == Solace.UserPropertyValue.Kind.NONE) { + LOG.warn("Unsupported user property type: {}. ", value.getClass()); + continue; + } + + userProperties.put(key, userPropertyValue); + } catch (SDTException e) { + throw new RuntimeException(e); + } + } + return Collections.unmodifiableMap(userProperties); + } + + public static SDTMap toSDTMap(@Nullable Map properties) { + SDTMap sdtMap = JCSMPFactory.onlyInstance().createMap(); + + if (properties == null) { + return sdtMap; + } + + for (Map.Entry entry : properties.entrySet()) { + try { + putUserProperty(sdtMap, entry.getKey(), entry.getValue()); + } catch (SDTException e) { + throw new RuntimeException(e); + } + } + return sdtMap; + } + + private static Solace.UserPropertyValue toUserPropertyValue(@NonNull Object value) { + if (value instanceof Boolean) { + return Solace.UserPropertyValue.of((Boolean) value); + } + if (value instanceof Byte) { + return Solace.UserPropertyValue.of((Byte) value); + } + if (value instanceof Short) { + return Solace.UserPropertyValue.of((Short) value); + } + if (value instanceof Integer) { + return Solace.UserPropertyValue.of((Integer) value); + } + if (value instanceof Long) { + return Solace.UserPropertyValue.of((Long) value); + } + if (value instanceof Float) { + return Solace.UserPropertyValue.of((Float) value); + } + if (value instanceof Double) { + return Solace.UserPropertyValue.of((Double) value); + } + if (value instanceof Character) { + return Solace.UserPropertyValue.of((Character) value); + } + if (value instanceof String) { + return Solace.UserPropertyValue.of((String) value); + } + if (value instanceof byte[]) { + return Solace.UserPropertyValue.of(Bytes.asList((byte[]) value)); + } + if (value instanceof ByteArray) { + return Solace.UserPropertyValue.of(Bytes.asList(((ByteArray) value).asBytes())); + } + if (value instanceof Destination) { + return Solace.UserPropertyValue.of(toSolaceDestination((Destination) value)); + } + return Solace.UserPropertyValue.of(); + } + + private static Solace.Destination toSolaceDestination(Destination destination) { + return Solace.Destination.builder() + .setType( + destination instanceof Topic + ? Solace.DestinationType.TOPIC + : Solace.DestinationType.QUEUE) + .setName(destination.getName()) + .build(); + } + + private static Destination toDestination(Solace.Destination destination) { + if (destination.getType() == Solace.DestinationType.QUEUE) { + return JCSMPFactory.onlyInstance().createQueue(destination.getName()); + } + return JCSMPFactory.onlyInstance().createTopic(destination.getName()); + } + + private static void putUserProperty( + SDTMap map, String key, Solace.UserPropertyValue propertyValue) + throws UnsupportedOperationException, SDTException { + if (propertyValue == null) { + return; + } + switch (propertyValue.getKind()) { + case BOOLEAN: + ifNotNull(propertyValue.getBoolean(), v -> map.putBoolean(key, v)); + return; + case BYTE: + ifNotNull(propertyValue.getByte(), v -> map.putByte(key, v)); + return; + case SHORT: + ifNotNull(propertyValue.getShort(), v -> map.putShort(key, v)); + return; + case INTEGER: + ifNotNull(propertyValue.getInteger(), v -> map.putInteger(key, v)); + return; + case LONG: + ifNotNull(propertyValue.getLong(), v -> map.putLong(key, v)); + return; + case FLOAT: + ifNotNull(propertyValue.getFloat(), v -> map.putFloat(key, v)); + return; + case DOUBLE: + ifNotNull(propertyValue.getDouble(), v -> map.putDouble(key, v)); + return; + case CHARACTER: + ifNotNull(propertyValue.getCharacter(), v -> map.putCharacter(key, v)); + return; + case STRING: + ifNotNull(propertyValue.getString(), v -> map.putString(key, v)); + return; + case BYTES: + ifNotNull(propertyValue.getBytes(), v -> map.putBytes(key, Bytes.toArray(v))); + return; + case DESTINATION: + ifNotNull(propertyValue.getDestination(), v -> map.putDestination(key, toDestination(v))); + return; + case NONE: + default: + } + } + + @FunctionalInterface + private interface SDTConsumer { + void accept(@NonNull T value) throws SDTException; + } + + private static void ifNotNull(@Nullable T value, SDTConsumer consumer) + throws SDTException { + if (value != null) { + consumer.accept(value); + } + } +} 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 15fe06103fbd..b3345dad6e0c 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,11 +22,16 @@ import com.solacesystems.jcsmp.BytesXMLMessage; import com.solacesystems.jcsmp.JCSMPFactory; import com.solacesystems.jcsmp.TextMessage; +import java.io.Serializable; import java.nio.ByteBuffer; import java.nio.charset.CharacterCodingException; import java.nio.charset.CodingErrorAction; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.apache.beam.sdk.io.solace.broker.SolaceUserPropertiesMapper; import org.apache.beam.sdk.schemas.AutoValueSchema; import org.apache.beam.sdk.schemas.annotations.DefaultSchema; import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; @@ -83,10 +88,135 @@ public enum DestinationType { UNKNOWN } + /** An immutable, typed value carried by a user-property map. */ + @AutoValue + @DefaultSchema(AutoValueSchema.class) + public abstract static class UserPropertyValue { + public enum Kind { + NONE, + BOOLEAN, + BYTE, + SHORT, + INTEGER, + LONG, + FLOAT, + DOUBLE, + CHARACTER, + STRING, + BYTES, + DESTINATION + } + + public abstract Kind getKind(); + + public abstract @Nullable Boolean getBoolean(); + + public abstract @Nullable Byte getByte(); + + public abstract @Nullable Short getShort(); + + public abstract @Nullable Integer getInteger(); + + public abstract @Nullable Long getLong(); + + public abstract @Nullable Float getFloat(); + + public abstract @Nullable Double getDouble(); + + public abstract @Nullable Character getCharacter(); + + public abstract @Nullable String getString(); + + public abstract @Nullable List getBytes(); + + public abstract @Nullable Destination getDestination(); + + public static UserPropertyValue of() { + return builder(Kind.NONE).build(); + } + + public static UserPropertyValue of(Boolean value) { + return builder(Kind.BOOLEAN).setBoolean(value).build(); + } + + public static UserPropertyValue of(Byte value) { + return builder(Kind.BYTE).setByte(value).build(); + } + + public static UserPropertyValue of(Short value) { + return builder(Kind.SHORT).setShort(value).build(); + } + + public static UserPropertyValue of(Integer value) { + return builder(Kind.INTEGER).setInteger(value).build(); + } + + public static UserPropertyValue of(Long value) { + return builder(Kind.LONG).setLong(value).build(); + } + + public static UserPropertyValue of(Float value) { + return builder(Kind.FLOAT).setFloat(value).build(); + } + + public static UserPropertyValue of(Double value) { + return builder(Kind.DOUBLE).setDouble(value).build(); + } + + public static UserPropertyValue of(Character value) { + return builder(Kind.CHARACTER).setCharacter(value).build(); + } + + public static UserPropertyValue of(String value) { + return builder(Kind.STRING).setString(value).build(); + } + + public static UserPropertyValue of(List value) { + return builder(Kind.BYTES).setBytes(List.copyOf(value)).build(); + } + + public static UserPropertyValue of(Destination destination) { + return builder(Kind.DESTINATION).setDestination(destination).build(); + } + + private static Builder builder(Kind kind) { + return new AutoValue_Solace_UserPropertyValue.Builder().setKind(kind); + } + + @AutoValue.Builder + abstract static class Builder { + abstract Builder setKind(Kind value); + + abstract Builder setBoolean(@Nullable Boolean value); + + abstract Builder setByte(@Nullable Byte value); + + abstract Builder setShort(@Nullable Short value); + + abstract Builder setInteger(@Nullable Integer value); + + abstract Builder setLong(@Nullable Long value); + + abstract Builder setFloat(@Nullable Float value); + + abstract Builder setDouble(@Nullable Double value); + + abstract Builder setCharacter(@Nullable Character value); + + abstract Builder setString(@Nullable String value); + + abstract Builder setBytes(@Nullable List value); + + abstract Builder setDestination(@Nullable Destination value); + + abstract UserPropertyValue build(); + } + } + /** Represents a Solace message destination (either a Topic or a Queue). */ @AutoValue @DefaultSchema(AutoValueSchema.class) - public abstract static class Destination { + public abstract static class Destination implements Serializable { /** * Gets the name of the destination. * @@ -276,6 +406,15 @@ public enum PayloadType { @SchemaFieldNumber("13") public abstract PayloadType getPayloadType(); + /** + * Gets the typed, SDK-independent user properties of the message. Non beam-schema compatible + * types are not supported (SDTMap and SDTStream) + * + * @return The user properties, or an empty map if the message carries none. + */ + @SchemaFieldNumber("14") + public abstract Map getUserProperties(); + /** Gets the payload decoded as UTF-8 when this record has type {@link PayloadType#TEXT}. */ public final String getText() { if (getPayloadType() != PayloadType.TEXT) { @@ -292,7 +431,8 @@ public static Builder builder() { .setRedelivered(false) .setTimeToLive(0) .setAttachmentBytes(new byte[0]) - .setPayloadType(PayloadType.BYTES_XML); + .setPayloadType(PayloadType.BYTES_XML) + .setUserProperties(Collections.emptyMap()); } @AutoValue.Builder @@ -332,6 +472,8 @@ public abstract Builder setReplicationGroupMessageId( public abstract Builder setAttachmentBytes(byte[] attachmentBytes); + public abstract Builder setUserProperties(Map userProperties); + public abstract Record build(); } @@ -456,6 +598,8 @@ public static class SolaceRecordMapper { Destination replyTo = getDestination(msg.getCorrelationId(), msg.getReplyTo()); Destination destination = getDestination(msg.getCorrelationId(), msg.getDestination()); + Map userProperties = + SolaceUserPropertiesMapper.toUserPropertyValueMap(msg.getProperties()); Record.Builder recordBuilder = decodePayload(msg); return recordBuilder @@ -473,6 +617,7 @@ public static class SolaceRecordMapper { msg.getReplicationGroupMessageId() != null ? msg.getReplicationGroupMessageId().toString() : null) + .setUserProperties(userProperties) .build(); } @@ -519,6 +664,10 @@ public static BytesXMLMessage toMessage(Record record) { msg.setSenderTimestamp(senderTimestamp); msg.setApplicationMessageId(record.getMessageId()); + if (!record.getUserProperties().isEmpty()) { + msg.setProperties(SolaceUserPropertiesMapper.toSDTMap(record.getUserProperties())); + } + return msg; } 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 cc6567b4c88c..99deac638ea5 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 @@ -22,16 +22,22 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import com.solacesystems.common.util.ByteArray; import com.solacesystems.jcsmp.BytesMessage; import com.solacesystems.jcsmp.BytesXMLMessage; import com.solacesystems.jcsmp.DeliveryMode; import com.solacesystems.jcsmp.JCSMPFactory; +import com.solacesystems.jcsmp.SDTException; +import com.solacesystems.jcsmp.SDTMap; import com.solacesystems.jcsmp.TextMessage; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; 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.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Bytes; import org.junit.Test; public class SolaceRecordMapperTest { @@ -144,6 +150,48 @@ public void testMapMessageMetadata() { assertEquals(789L, record.getTimeToLive()); } + @Test + public void testMapMessageUserProperties() throws SDTException { + BytesXMLMessage message = JCSMPFactory.onlyInstance().createBytesXMLMessage(); + message.setApplicationMessageId("id"); + SDTMap properties = JCSMPFactory.onlyInstance().createMap(); + properties.putString("string", "value"); + properties.putBoolean("boolean", true); + properties.putDouble("double", 1.23); + properties.putFloat("float", 4.56f); + properties.putLong("long", 123456789L); + properties.putInteger("integer", 3); + properties.putShort("short", (short) 123); + properties.putCharacter("character", 'c'); + properties.putByte("byte", (byte) 1); + properties.putBytes("bytes", new byte[] {1, 2, 3}); + properties.putByteArray("byteArray", new ByteArray(new byte[] {4, 5, 6})); + properties.putDestination("topic", JCSMPFactory.onlyInstance().createTopic("topic")); + properties.putDestination("queue", JCSMPFactory.onlyInstance().createQueue("queue")); + + // unsupported types will be ignored + properties.putMap("unsupported-map", JCSMPFactory.onlyInstance().createMap()); + properties.putStream("unsupported-stream", JCSMPFactory.onlyInstance().createStream()); + + message.setProperties(properties); + + Record record = Solace.SolaceRecordMapper.toRecord(message); + + properties.remove("unsupported-map"); + properties.remove("unsupported-stream"); + assertEquals(properties.keySet(), record.getUserProperties().keySet()); + } + + @Test + public void testMapWithEmptyMessageUserProperties() { + BytesXMLMessage message = JCSMPFactory.onlyInstance().createBytesXMLMessage(); + message.setApplicationMessageId("id"); + + Record record = Solace.SolaceRecordMapper.toRecord(message); + + assertTrue(record.getUserProperties().isEmpty()); + } + @Test public void testMapTextRecord() { Record record = @@ -257,6 +305,70 @@ public void testToMessageDoesNotSetPublishingFields() { assertNull(msg.getCorrelationKey()); } + @Test + public void testMapRecordUserProperties() throws Exception { + Map userProperties = new HashMap<>(); + userProperties.put("string", Solace.UserPropertyValue.of("value")); + userProperties.put("boolean", Solace.UserPropertyValue.of(true)); + userProperties.put("double", Solace.UserPropertyValue.of(1.23)); + userProperties.put("float", Solace.UserPropertyValue.of(4.56f)); + userProperties.put("long", Solace.UserPropertyValue.of(123456789L)); + userProperties.put("integer", Solace.UserPropertyValue.of(3)); + userProperties.put("short", Solace.UserPropertyValue.of((short) 123)); + userProperties.put("character", Solace.UserPropertyValue.of('c')); + userProperties.put("byte", Solace.UserPropertyValue.of((byte) 1)); + userProperties.put("bytes", Solace.UserPropertyValue.of(Bytes.asList(new byte[] {1, 2, 3}))); + userProperties.put( + "topic", + Solace.UserPropertyValue.of( + Solace.Destination.builder() + .setType(Solace.DestinationType.TOPIC) + .setName("topic") + .build())); + userProperties.put( + "queue", + Solace.UserPropertyValue.of( + Solace.Destination.builder() + .setType(Solace.DestinationType.QUEUE) + .setName("queue") + .build())); + + Record record = + Record.builder() + .setMessageId("id") + .setText("test") + .setUserProperties(userProperties) + .build(); + + BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record); + + SDTMap expected = JCSMPFactory.onlyInstance().createMap(); + expected.putString("string", "value"); + expected.putBoolean("boolean", true); + expected.putDouble("double", 1.23); + expected.putFloat("float", 4.56f); + expected.putLong("long", 123456789L); + expected.putInteger("integer", 3); + expected.putShort("short", (short) 123); + expected.putCharacter("character", 'c'); + expected.putByte("byte", (byte) 1); + expected.putBytes("bytes", new byte[] {1, 2, 3}); + expected.putDestination("topic", JCSMPFactory.onlyInstance().createTopic("topic")); + expected.putDestination("queue", JCSMPFactory.onlyInstance().createQueue("queue")); + + assertEquals(expected, msg.getProperties()); + } + + @Test + public void testMapWithEmptyRecordUserProperties() { + Record record = + Record.builder().setMessageId("id").setText("hello").setSenderTimestamp(1L).build(); + + BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record); + + assertNull(msg.getProperties()); + } + // --------------------------------------------------------------------------- // round-trip // --------------------------------------------------------------------------- @@ -310,4 +422,18 @@ public void testRoundTripBytesXmlPayloadWithAttachment() { assertArrayEquals(new byte[] {1, 2}, Arrays.copyOf(decoded.getPayload(), 2)); assertArrayEquals(new byte[] {3, 4}, decoded.getAttachmentBytes()); } + + private static void assertUserPropertiesEqual( + SDTMap expected, Map actual) throws SDTException { + assertEquals(expected.keySet(), actual.keySet()); + for (String key : expected.keySet()) { + Object expectedValue = expected.get(key); + Object actualValue = actual.get(key); + if (expectedValue instanceof byte[]) { + assertArrayEquals((byte[]) expectedValue, (byte[]) actualValue); + } else { + assertEquals(expectedValue, actualValue); + } + } + } }