Skip to content

feat(sdk-metrics): align PeriodicMetricReader export timeout semantics #8684

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

Open
Rajkaran-122 wants to merge 10 commits into
base: main
Choose a base branch
from
Open

feat(sdk-metrics): align PeriodicMetricReader export timeout semantics #8684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
41910f3
feat: align PeriodicMetricReader export timeout semantics
Rajkaran-122 Aug 2, 2026
e371867
fix: resolve test failures by implementing export timeout asynchronously
Rajkaran-122 Aug 2, 2026
d3fc20b
fix: use CompletableResultCode.join() for export timeout instead of s…
Rajkaran-122 Aug 2, 2026
b78cae9
fix: resolve errorprone warnings
Rajkaran-122 Aug 2, 2026
1b69dec
feat: align PeriodicMetricReader export timeout semantics with specif…
Rajkaran-122 Aug 21, 2026
c3b8e7f
fix: resolve JavaDoc formatting violations in PeriodicMetricReaderBui…
Rajkaran-122 Aug 21, 2026
7d07ef8
fix: apply Spotless formatting changes
Rajkaran-122 Aug 21, 2026
bc2f19d
feat: implement async exporter timeout enforcement for PeriodicMetric…
Rajkaran-122 Aug 21, 2026
b25a0bb
refactor: use 2-thread scheduler and simplify timeout enforcement per…
Rajkaran-122 Aug 21, 2026
acca5b7
Address review comments for exporter timeout defaults and fast-path o…
Rajkaran-122 Aug 22, 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
2 changes: 2 additions & 0 deletions docs/apidiffs/current_vs_latest/opentelemetry-sdk-metrics.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
Comparing source compatibility of opentelemetry-sdk-metrics-1.65.0-SNAPSHOT.jar against opentelemetry-sdk-metrics-1.64.0.jar
*** MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.sdk.metrics.export.PeriodicMetricReaderBuilder (not serializable)
=== CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.PeriodicMetricReaderBuilder setExporterTimeout(long, java.util.concurrent.TimeUnit)
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.PeriodicMetricReaderBuilder setExporterTimeout(java.time.Duration)
+++ NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.export.PeriodicMetricReaderBuilder setInternalTelemetryVersion(io.opentelemetry.sdk.common.InternalTelemetryVersion)
55 changes: 50 additions & 5 deletions sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import io.opentelemetry.sdk.metrics.data.MetricData;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -46,6 +47,7 @@ public final class PeriodicMetricReader implements MetricReader {

private final MetricExporter exporter;
private final long intervalNanos;
private final long exporterTimeoutNanos;
private final ScheduledExecutorService scheduler;
private final Scheduled scheduled;
private final Object lock = new Object();
Expand All @@ -72,11 +74,13 @@ public static PeriodicMetricReaderBuilder builder(MetricExporter exporter) {
PeriodicMetricReader(
MetricExporter exporter,
long intervalNanos,
long exporterTimeoutNanos,
ScheduledExecutorService scheduler,
int maxExportBatchSize,
InternalTelemetryVersion internalTelemetryVersion) {
this.exporter = exporter;
this.intervalNanos = intervalNanos;
this.exporterTimeoutNanos = exporterTimeoutNanos;
this.scheduler = scheduler;
this.maxExportBatchSize = maxExportBatchSize;
this.scheduled = new Scheduled();
Expand Down Expand Up @@ -213,7 +217,8 @@ private Scheduled() {}

private CompletableResultCode exportMetrics(Collection<MetricData> metricData) {
if (maxExportBatchSize == 0) {
return exporter.export(metricData);
CompletableResultCode result = exporter.export(metricData);
return applyTimeout(result);
}
Collection<Collection<MetricData>> batches =
MetricExportBatcher.batchMetrics(metricData, maxExportBatchSize);
Expand All @@ -227,14 +232,15 @@ public void run() {
while (batchIterator.hasNext()) {
Collection<MetricData> currentBatch = batchIterator.next();
CompletableResultCode currentResult = exporter.export(currentBatch);
if (currentResult.isDone()) {
if (!currentResult.isSuccess()) {
CompletableResultCode timeoutResult = applyTimeout(currentResult);
if (timeoutResult.isDone()) {
if (!timeoutResult.isSuccess()) {
anyFailed.set(true);
}
} else {
currentResult.whenComplete(
timeoutResult.whenComplete(
() -> {
if (!currentResult.isSuccess()) {
if (!timeoutResult.isSuccess()) {
anyFailed.set(true);
}
this.run();
Expand All @@ -253,6 +259,45 @@ public void run() {
return sequentialResult;
}

private CompletableResultCode applyTimeout(CompletableResultCode result) {
if (exporterTimeoutNanos == Long.MAX_VALUE) {
return result;
}

if (result.isDone()) {
return result;
}

try {
CompletableResultCode timeoutResult = new CompletableResultCode();

ScheduledFuture<?> timeoutFuture =
scheduler.schedule(
() -> {
logger.log(
Level.WARNING, "Export timed out after " + exporterTimeoutNanos + "ns");
timeoutResult.fail();
},
exporterTimeoutNanos,
TimeUnit.NANOSECONDS);

result.whenComplete(
() -> {
timeoutFuture.cancel(false);
if (result.isSuccess()) {
timeoutResult.succeed();
} else {
timeoutResult.fail();
}
});

return timeoutResult;
} catch (RejectedExecutionException e) {
// Scheduler is shutting down, return original result without timeout enforcement
return result;
}
}
Comment on lines +262 to +299

@jack-berg jack-berg Aug 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Suggested change
private CompletableResultCode applyTimeout(CompletableResultCode result) {
if (exporterTimeoutNanos == Long.MAX_VALUE) {
return result;
}
CompletableResultCode timeoutResult = new CompletableResultCode();
AtomicBoolean timedOut = new AtomicBoolean(false);
ScheduledFuture<?> timeoutFuture =
timeoutExecutor.schedule(
() -> {
if (!result.isDone()) {
timedOut.set(true);
logger.log(
Level.WARNING, "Export timed out after " + exporterTimeoutNanos + "ns");
timeoutResult.fail();
}
},
exporterTimeoutNanos,
TimeUnit.NANOSECONDS);
result.whenComplete(
() -> {
if (timeoutFuture != null) {
timeoutFuture.cancel(false);
}
if (!timedOut.get()) {
if (result.isSuccess()) {
timeoutResult.succeed();
} else {
timeoutResult.fail();
}
}
});
return timeoutResult;
}
private CompletableResultCode applyTimeout(CompletableResultCode result) {
if (exporterTimeoutNanos == Long.MAX_VALUE) {
return result;
}
CompletableResultCode timeoutResult = new CompletableResultCode();
ScheduledFuture<?> timeoutFuture =
scheduler.schedule(
() -> {
logger.log(Level.WARNING, "Export timed out after " + exporterTimeoutNanos + "ns");
timeoutResult.fail();
},
exporterTimeoutNanos,
TimeUnit.NANOSECONDS);
result.whenComplete(
() -> {
timeoutFuture.cancel(false);
if (result.isSuccess()) {
timeoutResult.succeed();
} else {
timeoutResult.fail();
}
});
return timeoutResult;
}


void setMeterProvider(MeterProvider meterProvider) {
instrumentation = new MetricReaderInstrumentation(COMPONENT_ID, meterProvider);
}
Expand Down
34 changes: 32 additions & 2 deletions ...etrics/src/main/java/io/opentelemetry/sdk/metrics/export/PeriodicMetricReaderBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,16 @@
public final class PeriodicMetricReaderBuilder {

static final long DEFAULT_SCHEDULE_DELAY_MINUTES = 1;
static final int DEFAULT_EXPORT_TIMEOUT_MILLIS = 30_000;

@jack-berg jack-berg Aug 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Adding a default timeout when none exists today is problematic, since it can change behavior that has been otherwise working. We either have to call the lack of a timeout a bug, or be more conservative and set the default timeout to the interval when the user doesn't explicitly set it.

@Rajkaran-122 Rajkaran-122 Aug 21, 2026 •
edited
Loading

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 @jack-berg sir or the feedback. I’ve addressed the API review comments, generated the API diff, and added tests for the new timeout API.

I also updated the timeout implementation to avoid blocking the PeriodicMetricReader scheduler while preserving asynchronous batching and handling shutdown/final export safely.

The full CI matrix is now passing. I’d appreciate your review of the updated implementation.

@jack-berg jack-berg Aug 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This comment is not addressed. To address, you would need to leave exporterTimeoutNanos as null, and resolve the value conditionally in build().


private final MetricExporter metricExporter;

private InternalTelemetryVersion internalTelemetryVersion = InternalTelemetryVersion.LATEST;

private long intervalNanos = TimeUnit.MINUTES.toNanos(DEFAULT_SCHEDULE_DELAY_MINUTES);

@Nullable private Long exporterTimeoutNanos;

@Nullable private ScheduledExecutorService executor;

private int maxExportBatchSize;
Expand All @@ -57,6 +60,26 @@ public PeriodicMetricReaderBuilder setInterval(Duration interval) {
return setInterval(interval.toNanos(), TimeUnit.NANOSECONDS);
}

/**
* Sets the timeout for the underlying exporter. If unset, defaults to {@value
* DEFAULT_EXPORT_TIMEOUT_MILLIS}ms.
*/
public PeriodicMetricReaderBuilder setExporterTimeout(long timeout, TimeUnit unit) {

@jack-berg jack-berg Aug 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This is new public API surface area, which comes with new content checks into /docs/apidiffs. Please run the build to generate this.

Also, you'll see failing tests if you run the build. You'll want to fix those failing tests, and add new tests for this specific feature.

requireNonNull(unit, "unit");
checkArgument(timeout >= 0, "timeout must be non-negative");
exporterTimeoutNanos = timeout == 0 ? Long.MAX_VALUE : unit.toNanos(timeout);
return this;
}

/**
* Sets the timeout for the underlying exporter. If unset, defaults to {@value
* DEFAULT_EXPORT_TIMEOUT_MILLIS}ms.
*/
public PeriodicMetricReaderBuilder setExporterTimeout(Duration timeout) {
requireNonNull(timeout, "timeout");
return setExporterTimeout(timeout.toNanos(), TimeUnit.NANOSECONDS);
}

/** Sets the {@link ScheduledExecutorService} to schedule reads on. */
public PeriodicMetricReaderBuilder setExecutor(ScheduledExecutorService executor) {
requireNonNull(executor, "executor");
Expand All @@ -83,10 +106,17 @@ public PeriodicMetricReader build() {
ScheduledExecutorService executor = this.executor;
if (executor == null) {
executor =
Executors.newScheduledThreadPool(1, new DaemonThreadFactory("PeriodicMetricReader"));
Executors.newScheduledThreadPool(2, new DaemonThreadFactory("PeriodicMetricReader"));
}
return new PeriodicMetricReader(
metricExporter, intervalNanos, executor, maxExportBatchSize, internalTelemetryVersion);
metricExporter,
intervalNanos,
exporterTimeoutNanos != null
? exporterTimeoutNanos
: Math.min(intervalNanos, TimeUnit.MILLISECONDS.toNanos(DEFAULT_EXPORT_TIMEOUT_MILLIS)),
executor,
maxExportBatchSize,
internalTelemetryVersion);
}

/** Sets the internal telemetry version used to control self-observability metrics. */
Expand Down
Loading
Loading