Skip to content

First pass at flaky VertxReactivePropagationTest.highConcurrency #18511

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
trask wants to merge 4 commits into from
Closed

First pass at flaky VertxReactivePropagationTest.highConcurrency #18511

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
e41dcb6
First pass at flaky test
trask May 2, 2026
c0c9a43
spotless
trask May 2, 2026
767baa8
Merge remote-tracking branch 'upstream/main' into alt
trask May 2, 2026
b355146
review
trask May 11, 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
48 changes: 29 additions & 19 deletions ...elemetry/javaagent/instrumentation/vertx/rx/v3_5/server/VertxReactivePropagationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_SYSTEM;
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_USER;
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DbSystemNameIncubatingValues.HSQLDB;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.assertj.core.api.Assertions.assertThat;

import io.opentelemetry.api.GlobalOpenTelemetry;
Expand All @@ -59,6 +61,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import org.junit.jupiter.api.BeforeAll;
Expand Down Expand Up @@ -149,7 +152,7 @@ void contextPropagation() {

@SuppressWarnings("deprecation") // uses deprecated db semconv
@Test
void highConcurrency() {
void highConcurrency() throws Exception {
int count = 100;
String baseUrl = "/listProducts";
CountDownLatch latch = new CountDownLatch(1);
Expand All @@ -160,29 +163,36 @@ void highConcurrency() {
TextMapSetter<HttpRequestBuilder> setter =
(carrier, name, value) -> carrier.header(name, value);

List<Future<?>> futures = new ArrayList<>();

@laurit laurit May 12, 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.

I guess this won't do anything to fix the flaky test. If you believe that waiting for all the submitted tasks helps I'd suggest to also change the highConcurrency method in AbstractHttpClientTest that these methods copy from.
I get why copilot is forcing recomputing the deadline, but it looks a bit too much effort. Maybe we could avoid it with the following

CompletableFuture<?>[] tasks = new CompletableFuture[count];
for (int i = 0; i < count; i++) {
  tasks[i] = CompletableFuture.runAsync(job, pool);
}
latch.countDown();
CompletableFuture.allOf(tasks).get(30, SECONDS);

for (int i = 0; i < count; i++) {
int index = i;
pool.submit(
() -> {
try {
latch.await();
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
testing.runWithSpan(
"client " + index,
() -> {
HttpRequestBuilder builder =
HttpRequest.builder()
.get(baseUrl + "?" + TEST_REQUEST_ID_PARAMETER + "=" + index);
Span.current().setAttribute(TEST_REQUEST_ID_ATTRIBUTE, index);
propagator.inject(Context.current(), builder, setter);
client.execute(builder.build()).aggregate().join();
});
});
futures.add(
pool.submit(
() -> {
try {
assertThat(latch.await(10, SECONDS)).isTrue();
Comment thread
trask marked this conversation as resolved.
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new AssertionError(e);
}
testing.runWithSpan(
"client " + index,
() -> {
HttpRequestBuilder builder =
HttpRequest.builder()
.get(baseUrl + "?" + TEST_REQUEST_ID_PARAMETER + "=" + index);
Span.current().setAttribute(TEST_REQUEST_ID_ATTRIBUTE, index);
propagator.inject(Context.current(), builder, setter);
client.execute(builder.build()).aggregate().join();
});
}));
}

latch.countDown();
long deadlineNanos = System.nanoTime() + SECONDS.toNanos(30);
for (Future<?> future : futures) {
future.get(Math.max(deadlineNanos - System.nanoTime(), 0), NANOSECONDS);
}
Comment thread
trask marked this conversation as resolved.
Comment thread
trask marked this conversation as resolved.
Comment thread
trask marked this conversation as resolved.

List<Consumer<TraceAssert>> assertions = new ArrayList<>();
for (int i = 0; i < count; i++) {
Expand Down
48 changes: 29 additions & 19 deletions ...elemetry/javaagent/instrumentation/vertx/rx/v3_5/server/VertxReactivePropagationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_SYSTEM;
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_USER;
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DbSystemNameIncubatingValues.HSQLDB;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.assertj.core.api.Assertions.assertThat;

import io.opentelemetry.api.GlobalOpenTelemetry;
Expand All @@ -59,6 +61,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import org.junit.jupiter.api.BeforeAll;
Expand Down Expand Up @@ -149,7 +152,7 @@ void contextPropagation() {

@SuppressWarnings("deprecation") // uses deprecated db semconv
@Test
void highConcurrency() {
void highConcurrency() throws Exception {
int count = 100;
String baseUrl = "/listProducts";
CountDownLatch latch = new CountDownLatch(1);
Expand All @@ -160,29 +163,36 @@ void highConcurrency() {
TextMapSetter<HttpRequestBuilder> setter =
(carrier, name, value) -> carrier.header(name, value);

List<Future<?>> futures = new ArrayList<>();
for (int i = 0; i < count; i++) {
int index = i;
pool.submit(
() -> {
try {
latch.await();
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
testing.runWithSpan(
"client " + index,
() -> {
HttpRequestBuilder builder =
HttpRequest.builder()
.get(baseUrl + "?" + TEST_REQUEST_ID_PARAMETER + "=" + index);
Span.current().setAttribute(TEST_REQUEST_ID_ATTRIBUTE, index);
propagator.inject(Context.current(), builder, setter);
client.execute(builder.build()).aggregate().join();
});
});
futures.add(
pool.submit(
() -> {
try {
assertThat(latch.await(10, SECONDS)).isTrue();
Comment thread
trask marked this conversation as resolved.
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new AssertionError(e);
}
testing.runWithSpan(
"client " + index,
() -> {
HttpRequestBuilder builder =
HttpRequest.builder()
.get(baseUrl + "?" + TEST_REQUEST_ID_PARAMETER + "=" + index);
Span.current().setAttribute(TEST_REQUEST_ID_ATTRIBUTE, index);
propagator.inject(Context.current(), builder, setter);
client.execute(builder.build()).aggregate().join();
});
}));
}

latch.countDown();
long deadlineNanos = System.nanoTime() + SECONDS.toNanos(30);
for (Future<?> future : futures) {
future.get(Math.max(deadlineNanos - System.nanoTime(), 0), NANOSECONDS);
}

List<Consumer<TraceAssert>> assertions = new ArrayList<>();
for (int i = 0; i < count; i++) {
Expand Down
48 changes: 29 additions & 19 deletions ...elemetry/javaagent/instrumentation/vertx/rx/v3_5/server/VertxReactivePropagationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_SYSTEM;
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_USER;
import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DbSystemNameIncubatingValues.HSQLDB;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.assertj.core.api.Assertions.assertThat;

import io.opentelemetry.api.GlobalOpenTelemetry;
Expand All @@ -59,6 +61,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import org.junit.jupiter.api.BeforeAll;
Expand Down Expand Up @@ -149,7 +152,7 @@ void contextPropagation() {

@SuppressWarnings("deprecation") // uses deprecated db semconv
@Test
void highConcurrency() {
void highConcurrency() throws Exception {
int count = 100;
String baseUrl = "/listProducts";
CountDownLatch latch = new CountDownLatch(1);
Expand All @@ -160,29 +163,36 @@ void highConcurrency() {
TextMapSetter<HttpRequestBuilder> setter =
(carrier, name, value) -> carrier.header(name, value);

List<Future<?>> futures = new ArrayList<>();
for (int i = 0; i < count; i++) {
int index = i;
pool.submit(
() -> {
try {
latch.await();
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
testing.runWithSpan(
"client " + index,
() -> {
HttpRequestBuilder builder =
HttpRequest.builder()
.get(baseUrl + "?" + TEST_REQUEST_ID_PARAMETER + "=" + index);
Span.current().setAttribute(TEST_REQUEST_ID_ATTRIBUTE, index);
propagator.inject(Context.current(), builder, setter);
client.execute(builder.build()).aggregate().join();
});
});
futures.add(
pool.submit(
() -> {
try {
assertThat(latch.await(10, SECONDS)).isTrue();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new AssertionError(e);
Comment thread
trask marked this conversation as resolved.
}
testing.runWithSpan(
"client " + index,
() -> {
HttpRequestBuilder builder =
HttpRequest.builder()
.get(baseUrl + "?" + TEST_REQUEST_ID_PARAMETER + "=" + index);
Span.current().setAttribute(TEST_REQUEST_ID_ATTRIBUTE, index);
propagator.inject(Context.current(), builder, setter);
client.execute(builder.build()).aggregate().join();
});
}));
Comment thread
trask marked this conversation as resolved.
}

latch.countDown();
long deadlineNanos = System.nanoTime() + SECONDS.toNanos(30);
for (Future<?> future : futures) {
future.get(Math.max(deadlineNanos - System.nanoTime(), 0), NANOSECONDS);
}

List<Consumer<TraceAssert>> assertions = new ArrayList<>();
for (int i = 0; i < count; i++) {
Expand Down
Loading