Skip to content

Add OpenTelemetry tracing support to SolaceIO #40187

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
singhmrinal02 wants to merge 4 commits into from
Closed

Add OpenTelemetry tracing support to SolaceIO #40187

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
efa8443
Fix JaCoCo coverage for multiple test tasks
singhmrinal02 Sep 17, 2026
cf07ab2
Add OpenTelemetry tracing support to SolaceIO
singhmrinal02 Sep 20, 2026
764ed67
Resolve Solace OpenTelemetry rebase conflict
singhmrinal02 Sep 20, 2026
362c1a4
Add validation for Solace OpenTelemetry tracing
singhmrinal02 Sep 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1418,9 +1418,10 @@ class BeamModulePlugin implements Plugin<Project> {
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
Expand Down
54 changes: 38 additions & 16 deletions sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -472,7 +473,8 @@ public static Read<Solace.Record> 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));
}

/**
Expand Down Expand Up @@ -503,7 +505,8 @@ public static <T> Read<T> 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));
}

/**
Expand Down Expand Up @@ -627,6 +630,11 @@ public Read<T> withDeduplicateRecords(boolean deduplicateRecords) {
return this;
}

public Read<T> withEnableOpenTelemetryTracing() {
configurationBuilder.setEnableOpenTelemetryTracing(true);

@stankiewicz stankiewicz Sep 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add check.
This feature will only work if TypeDescriptor.of(Solace.Record.class) , ParDo after UnboundedSource need to access headers, which are available in Solace.Record.

@singhmrinal02 singhmrinal02 Sep 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for reviewing and for the guidance! I’ll check again and work on the missing parts.

return this;
}

/**
* Set a factory that creates a {@link org.apache.beam.sdk.io.solace.broker.SempClientFactory}.
*
Expand Down Expand Up @@ -733,6 +741,8 @@ abstract static class Configuration<T> {

abstract boolean getNackOnTimeout();

abstract boolean getEnableOpenTelemetryTracing();

public static <T> Builder<T> builder() {
Builder<T> builder =
new org.apache.beam.sdk.io.solace.AutoValue_SolaceIO_Read_Configuration.Builder<T>();
Expand All @@ -752,6 +762,8 @@ public abstract static class Builder<T> {

abstract Builder<T> setDeduplicateRecords(boolean deduplicateRecords);

abstract Builder<T> setEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing);

abstract Builder<T> setParseFn(
SerializableFunction<@Nullable BytesXMLMessage, @Nullable T> parseFn);

Expand All @@ -777,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
Expand All @@ -793,20 +811,24 @@ public PCollection<T> expand(PBegin input) {

Coder<T> 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<T> 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
Expand Down
5 changes: 5 additions & 0 deletions ...ava/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ class UnboundedSolaceReader<T> extends UnboundedReader<T> {

private final Duration ackDeadline;
private final boolean nackOnTimeout;
private final boolean enableOpenTelemetryTracing;

/**
* Map to track pending checkpoints and their messages. Accessed by both reader
Expand Down Expand Up @@ -135,6 +136,7 @@ public UnboundedSolaceReader(UnboundedSolaceSource<T> 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() {
Expand Down Expand Up @@ -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();
Expand Down
12 changes: 10 additions & 2 deletions ...ava/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public class UnboundedSolaceSource<T> extends UnboundedSource<T, SolaceCheckpoin
private final SerializableFunction<@Nullable BytesXMLMessage, @Nullable T> parseFn;
private final Duration ackDeadline;
private final boolean nackOnTimeout;
private final boolean enableOpenTelemetryTracing;

public Queue getQueue() {
return queue;
Expand Down Expand Up @@ -84,6 +85,10 @@ public boolean getNackOnTimeout() {
return nackOnTimeout;
}

public boolean getEnableOpenTelemetryTracing() {
return enableOpenTelemetryTracing;
}

public UnboundedSolaceSource(
Queue queue,
SempClientFactory sempClientFactory,
Expand All @@ -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;
Expand All @@ -107,6 +113,7 @@ public UnboundedSolaceSource(
this.parseFn = parseFn;
this.ackDeadline = ackDeadline;
this.nackOnTimeout = nackOnTimeout;
this.enableOpenTelemetryTracing = enableOpenTelemetryTracing;
}

@Override
Expand Down Expand Up @@ -152,7 +159,8 @@ private List<UnboundedSolaceSource<T>> getSolaceSources(
watermarkIdleDurationThreshold,
parseFn,
ackDeadline,
nackOnTimeout);
nackOnTimeout,
enableOpenTelemetryTracing);
sourceList.add(source);
}
return sourceList;
Expand Down
6 changes: 4 additions & 2 deletions sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,8 @@ private static UnboundedSolaceSource<Record> getSource(Read<Record> spec, TestPi
configuration.getWatermarkIdleDurationThreshold(),
configuration.getParseFn(),
configuration.getAckDeadline(),
configuration.getNackOnTimeout());
configuration.getNackOnTimeout(),
configuration.getEnableOpenTelemetryTracing());
}

@Test
Expand Down Expand Up @@ -542,7 +543,8 @@ public void testCheckpointMarkSafety() throws Exception {
@Test
public void testDefaultCoder() {
Coder<SolaceCheckpointMark> 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);
}
Expand Down
9 changes: 6 additions & 3 deletions ...io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Solace.Record> reader =
Expand Down Expand Up @@ -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<Solace.Record> reader =
Expand Down Expand Up @@ -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<Solace.Record> reader =
Expand Down
Loading