From efa84431f3475d461734a1f1e07654dbe15197d1 Mon Sep 17 00:00:00 2001 From: singhmrinal02 Date: Fri, 18 Sep 2026 02:56:48 +0530 Subject: [PATCH 1/4] Fix JaCoCo coverage for multiple test tasks --- .../groovy/org/apache/beam/gradle/BeamModulePlugin.groovy | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index 88c998ed9105..c80f5e2701e6 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -1418,9 +1418,10 @@ class BeamModulePlugin implements Plugin { getSourceDirectories().setFrom( project.files(project.sourceSets.main.allSource.srcDirs) ) - getExecutionData().setFrom(project.file( - project.getLayout().getBuildDirectory().file("jacoco/test.exec") - )) + getExecutionData().setFrom(project.fileTree( + dir: project.getLayout().getBuildDirectory().dir("jacoco"), + include: ['*.exec'] + )) reports { html.required = true xml.required = true From cf07ab210a912b22205afa7c69b0b7e40ea6a2a7 Mon Sep 17 00:00:00 2001 From: singhmrinal02 Date: Sun, 20 Sep 2026 12:02:20 +0530 Subject: [PATCH 2/4] Add OpenTelemetry tracing support to SolaceIO B --- .../apache/beam/sdk/io/solace/SolaceIO.java | 48 ++++++++++++------- .../beam/sdk/io/solace/data/Solace.java | 26 +++++++++- .../io/solace/read/UnboundedSolaceReader.java | 5 ++ .../io/solace/read/UnboundedSolaceSource.java | 12 ++++- .../beam/sdk/io/solace/SolaceIOReadTest.java | 6 ++- .../read/UnboundedSolaceReaderTest.java | 9 ++-- 6 files changed, 81 insertions(+), 25 deletions(-) diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java index 57f65ae62651..4a3e79cff0f7 100644 --- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java +++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java @@ -423,6 +423,7 @@ public class SolaceIO { Duration.standardSeconds(30); private static final Duration DEFAULT_ACK_DEADLINE = Duration.standardSeconds(30); public static final boolean DEFAULT_NACK_ON_TIMEOUT = false; + private static final boolean DEFAULT_ENABLE_OPEN_TELEMETRY_TRACING = false; public static final int DEFAULT_WRITER_NUM_SHARDS = 20; public static final int DEFAULT_WRITER_CLIENTS_PER_WORKER = 4; public static final Boolean DEFAULT_WRITER_PUBLISH_LATENCY_METRICS = false; @@ -472,7 +473,8 @@ public static Read read() { .setDeduplicateRecords(DEFAULT_DEDUPLICATE_RECORDS) .setWatermarkIdleDurationThreshold(DEFAULT_WATERMARK_IDLE_DURATION_THRESHOLD) .setAckDeadline(DEFAULT_ACK_DEADLINE) - .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT)); + .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT) + .setEnableOpenTelemetryTracing(DEFAULT_ENABLE_OPEN_TELEMETRY_TRACING)); } /** @@ -503,7 +505,8 @@ public static Read read( .setDeduplicateRecords(DEFAULT_DEDUPLICATE_RECORDS) .setWatermarkIdleDurationThreshold(DEFAULT_WATERMARK_IDLE_DURATION_THRESHOLD) .setAckDeadline(DEFAULT_ACK_DEADLINE) - .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT)); + .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT) + .setEnableOpenTelemetryTracing(DEFAULT_ENABLE_OPEN_TELEMETRY_TRACING)); } /** @@ -627,6 +630,11 @@ public Read withDeduplicateRecords(boolean deduplicateRecords) { return this; } + public Read withEnableOpenTelemetryTracing() { + configurationBuilder.setEnableOpenTelemetryTracing(true); + return this; + } + /** * Set a factory that creates a {@link org.apache.beam.sdk.io.solace.broker.SempClientFactory}. * @@ -733,6 +741,8 @@ abstract static class Configuration { abstract boolean getNackOnTimeout(); + abstract boolean getEnableOpenTelemetryTracing(); + public static Builder builder() { Builder builder = new org.apache.beam.sdk.io.solace.AutoValue_SolaceIO_Read_Configuration.Builder(); @@ -752,6 +762,8 @@ public abstract static class Builder { abstract Builder setDeduplicateRecords(boolean deduplicateRecords); + abstract Builder setEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing); + abstract Builder setParseFn( SerializableFunction<@Nullable BytesXMLMessage, @Nullable T> parseFn); @@ -793,20 +805,24 @@ public PCollection expand(PBegin input) { Coder coder = inferCoder(input.getPipeline(), configuration.getTypeDescriptor()); - return input.apply( - org.apache.beam.sdk.io.Read.from( - new UnboundedSolaceSource<>( - initializedQueue, - sempClientFactory, - sessionServiceFactory, - configuration.getMaxNumConnections(), - configuration.getDeduplicateRecords(), - coder, - configuration.getTimestampFn(), - configuration.getWatermarkIdleDurationThreshold(), - configuration.getParseFn(), - configuration.getAckDeadline(), - configuration.getNackOnTimeout()))); + PCollection output = + input.apply( + org.apache.beam.sdk.io.Read.from( + new UnboundedSolaceSource<>( + initializedQueue, + sempClientFactory, + sessionServiceFactory, + configuration.getMaxNumConnections(), + configuration.getDeduplicateRecords(), + coder, + configuration.getTimestampFn(), + configuration.getWatermarkIdleDurationThreshold(), + configuration.getParseFn(), + configuration.getAckDeadline(), + configuration.getNackOnTimeout(), + configuration.getEnableOpenTelemetryTracing()))); + + return output; } @VisibleForTesting 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 b3345dad6e0c..8d85727207c0 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 @@ -21,6 +21,8 @@ import com.solacesystems.jcsmp.BytesMessage; import com.solacesystems.jcsmp.BytesXMLMessage; import com.solacesystems.jcsmp.JCSMPFactory; +import com.solacesystems.jcsmp.SDTException; +import com.solacesystems.jcsmp.SDTMap; import com.solacesystems.jcsmp.TextMessage; import java.io.Serializable; import java.nio.ByteBuffer; @@ -29,9 +31,11 @@ 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 java.util.HashMap; +import java.util.Map; import org.apache.beam.sdk.schemas.AutoValueSchema; import org.apache.beam.sdk.schemas.annotations.DefaultSchema; import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; @@ -414,7 +418,7 @@ public enum PayloadType { */ @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) { @@ -443,6 +447,8 @@ public abstract static class Builder { public abstract Builder setPayloadType(PayloadType payloadType); + public abstract Builder setProperties(Map properties); + /** 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); @@ -601,6 +607,22 @@ public static class SolaceRecordMapper { Map userProperties = SolaceUserPropertiesMapper.toUserPropertyValueMap(msg.getProperties()); + Map properties = new HashMap<>(); + SDTMap solaceProperties = msg.getProperties(); + if (solaceProperties != null) { + for (String key : solaceProperties.keySet()) { + Object value; + try { + value = solaceProperties.get(key); + } catch (SDTException e) { + continue; + } + if (value != null) { + properties.put(key, value.toString()); + } + } + } + Record.Builder recordBuilder = decodePayload(msg); return recordBuilder .setMessageId(msg.getApplicationMessageId()) diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java index 0bfd9f1c45d9..d4604de1f1eb 100644 --- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java +++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java @@ -80,6 +80,7 @@ class UnboundedSolaceReader extends UnboundedReader { private final Duration ackDeadline; private final boolean nackOnTimeout; + private final boolean enableOpenTelemetryTracing; /** * Map to track pending checkpoints and their messages. Accessed by both reader @@ -135,6 +136,7 @@ public UnboundedSolaceReader(UnboundedSolaceSource currentSource) { this.ackExecutor = Executors.newFixedThreadPool(4); this.ackDeadline = java.time.Duration.ofMillis(currentSource.getAckDeadline().getMillis()); this.nackOnTimeout = currentSource.getNackOnTimeout(); + this.enableOpenTelemetryTracing = currentSource.getEnableOpenTelemetryTracing(); } private SessionService getSessionService() { @@ -175,6 +177,9 @@ public boolean advance() { return false; } solaceOriginalRecord = receivedXmlMessage; + if (enableOpenTelemetryTracing) { + LOG.debug("SolaceIO.Read: OpenTelemetry tracing is enabled for the received message."); + } solaceMappedRecord = getCurrentSource().getParseFn().apply(receivedXmlMessage); receivedMessages.add(receivedXmlMessage); messagesReceived.inc(); diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java index 0ca9bff9ebf9..15fe46f9f078 100644 --- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java +++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java @@ -51,6 +51,7 @@ public class UnboundedSolaceSource extends UnboundedSource parseFn; private final Duration ackDeadline; private final boolean nackOnTimeout; + private final boolean enableOpenTelemetryTracing; public Queue getQueue() { return queue; @@ -84,6 +85,10 @@ public boolean getNackOnTimeout() { return nackOnTimeout; } + public boolean getEnableOpenTelemetryTracing() { + return enableOpenTelemetryTracing; + } + public UnboundedSolaceSource( Queue queue, SempClientFactory sempClientFactory, @@ -95,7 +100,8 @@ public UnboundedSolaceSource( Duration watermarkIdleDurationThreshold, SerializableFunction<@Nullable BytesXMLMessage, @Nullable T> parseFn, Duration ackDeadline, - boolean nackOnTimeout) { + boolean nackOnTimeout, + boolean enableOpenTelemetryTracing) { this.queue = queue; this.sempClientFactory = sempClientFactory; this.sessionServiceFactory = sessionServiceFactory; @@ -107,6 +113,7 @@ public UnboundedSolaceSource( this.parseFn = parseFn; this.ackDeadline = ackDeadline; this.nackOnTimeout = nackOnTimeout; + this.enableOpenTelemetryTracing = enableOpenTelemetryTracing; } @Override @@ -152,7 +159,8 @@ private List> getSolaceSources( watermarkIdleDurationThreshold, parseFn, ackDeadline, - nackOnTimeout); + nackOnTimeout, + enableOpenTelemetryTracing); sourceList.add(source); } return sourceList; diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java index 29e82142f94f..9ffe98bacfb4 100644 --- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java +++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java @@ -99,7 +99,8 @@ private static UnboundedSolaceSource getSource(Read spec, TestPi configuration.getWatermarkIdleDurationThreshold(), configuration.getParseFn(), configuration.getAckDeadline(), - configuration.getNackOnTimeout()); + configuration.getNackOnTimeout(), + configuration.getEnableOpenTelemetryTracing()); } @Test @@ -542,7 +543,8 @@ public void testCheckpointMarkSafety() throws Exception { @Test public void testDefaultCoder() { Coder coder = - new UnboundedSolaceSource<>(null, null, null, 0, false, null, null, null, null, null, false) + new UnboundedSolaceSource<>( + null, null, null, 0, false, null, null, null, null, null, false, false) .getCheckpointMarkCoder(); CoderProperties.coderSerializable(coder); } diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java index 33d463201a5e..8a62b23cc7f3 100644 --- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java +++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java @@ -67,7 +67,8 @@ public void testCheckpointTimeoutDefault_EvictsWithoutNack() throws Exception { org.joda.time.Duration.standardSeconds(1), // watermarkIdleDurationThreshold input -> SolaceDataUtils.getSolaceRecord("payload_test0", "450"), // parseFn org.joda.time.Duration.standardSeconds(30), // ackDeadline - false // nackOnTimeout (default) + false, // nackOnTimeout (default) + false // enableOpenTelemetryTracing (default) ); UnboundedSolaceReader reader = @@ -122,7 +123,8 @@ public void testCheckpointTimeout_WithNackEnabled() throws Exception { org.joda.time.Duration.standardSeconds(1), // watermarkIdleDurationThreshold input -> SolaceDataUtils.getSolaceRecord("payload_test0", "450"), // parseFn org.joda.time.Duration.standardSeconds(30), // ackDeadline - true // nackOnTimeout + true, // nackOnTimeout + false // enableOpenTelemetryTracing (default) ); UnboundedSolaceReader reader = @@ -188,7 +190,8 @@ public void testCheckpointTimeout_NackExceptionHandledGracefully() throws Except org.joda.time.Duration.standardSeconds(1), // watermarkIdleDurationThreshold input -> SolaceDataUtils.getSolaceRecord("payload_test0", "450"), // parseFn org.joda.time.Duration.standardSeconds(30), // ackDeadline - true // nackOnTimeout + true, // nackOnTimeout + false // enableOpenTelemetryTracing (default) ); UnboundedSolaceReader reader = From 764ed6734c8975a8965d8dbe39a8ff926562f2ee Mon Sep 17 00:00:00 2001 From: singhmrinal02 Date: Sun, 20 Sep 2026 13:15:52 +0530 Subject: [PATCH 3/4] Resolve Solace OpenTelemetry rebase conflict --- .../beam/sdk/io/solace/data/Solace.java | 26 ++----------------- 1 file changed, 2 insertions(+), 24 deletions(-) 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 8d85727207c0..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 @@ -21,8 +21,6 @@ import com.solacesystems.jcsmp.BytesMessage; import com.solacesystems.jcsmp.BytesXMLMessage; import com.solacesystems.jcsmp.JCSMPFactory; -import com.solacesystems.jcsmp.SDTException; -import com.solacesystems.jcsmp.SDTMap; import com.solacesystems.jcsmp.TextMessage; import java.io.Serializable; import java.nio.ByteBuffer; @@ -31,11 +29,9 @@ import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.Collections; - import java.util.List; -import org.apache.beam.sdk.io.solace.broker.SolaceUserPropertiesMapper; -import java.util.HashMap; 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; @@ -418,7 +414,7 @@ public enum PayloadType { */ @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) { @@ -447,8 +443,6 @@ public abstract static class Builder { public abstract Builder setPayloadType(PayloadType payloadType); - public abstract Builder setProperties(Map properties); - /** 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); @@ -607,22 +601,6 @@ public static class SolaceRecordMapper { Map userProperties = SolaceUserPropertiesMapper.toUserPropertyValueMap(msg.getProperties()); - Map properties = new HashMap<>(); - SDTMap solaceProperties = msg.getProperties(); - if (solaceProperties != null) { - for (String key : solaceProperties.keySet()) { - Object value; - try { - value = solaceProperties.get(key); - } catch (SDTException e) { - continue; - } - if (value != null) { - properties.put(key, value.toString()); - } - } - } - Record.Builder recordBuilder = decodePayload(msg); return recordBuilder .setMessageId(msg.getApplicationMessageId()) From 362c1a48f48ad105ab1a5df2656cac2bc2d1d7c5 Mon Sep 17 00:00:00 2001 From: singhmrinal02 Date: Thu, 24 Sep 2026 13:53:40 +0530 Subject: [PATCH 4/4] Add validation for Solace OpenTelemetry tracing --- .../main/java/org/apache/beam/sdk/io/solace/SolaceIO.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java index 4a3e79cff0f7..74ba8fa5a747 100644 --- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java +++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java @@ -789,6 +789,12 @@ public void validate(@Nullable PipelineOptions options) { checkState( (configuration.getQueue() == null ^ configuration.getTopic() == null), "SolaceIO.Read: One of the Solace {Queue, Topic} must be set."); + + if (configuration.getEnableOpenTelemetryTracing()) { + checkState( + TypeDescriptor.of(Solace.Record.class).isSupertypeOf(configuration.getTypeDescriptor()), + "OpenTelemetry tracing is only supported when the output type is Solace.Record."); + } } @Override