diff --git a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpClientTest.groovy b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpClientTest.groovy
index e787feeda1f..149ee9303f0 100644
--- a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpClientTest.groovy
+++ b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpClientTest.groovy
@@ -203,7 +203,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
     }
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -249,7 +249,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -300,7 +300,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -337,7 +337,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -376,7 +376,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -416,7 +416,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -451,7 +451,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -486,7 +486,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -534,7 +534,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -584,7 +584,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -624,7 +624,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -662,7 +662,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -726,7 +726,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -820,7 +820,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -850,7 +850,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
     }
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -885,7 +885,7 @@ abstract class HttpClientTest extends VersionedNamingTestBase {
     }
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
diff --git a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy
index f754a57e797..e227b22950c 100644
--- a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy
+++ b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy
@@ -684,7 +684,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     }
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -729,7 +729,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     }
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -776,7 +776,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     }
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -821,7 +821,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -886,7 +886,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -936,7 +936,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -980,7 +980,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1036,7 +1036,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1085,7 +1085,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1117,7 +1117,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1162,7 +1162,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1211,7 +1211,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1255,7 +1255,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1302,7 +1302,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1346,7 +1346,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1389,7 +1389,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1431,7 +1431,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1491,7 +1491,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1539,7 +1539,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1573,7 +1573,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1607,7 +1607,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1642,7 +1642,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1823,7 +1823,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1875,7 +1875,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1922,7 +1922,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -1962,7 +1962,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -2001,7 +2001,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -2063,7 +2063,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
@@ -2121,7 +2121,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
 
     and:
     if (isDataStreamsEnabled()) {
-      StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find {
+      StatsGroup first = TEST_DATA_STREAMS_WRITER.waitForGroup {
         it.parentHash == 0
       }
       verifyAll(first) {
diff --git a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriter.groovy b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriter.groovy
index 9feb48f3a98..2b79aef9b6e 100644
--- a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriter.groovy
+++ b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriter.groovy
@@ -87,7 +87,22 @@ class RecordingDatastreamsPayloadWriter implements DatastreamsPayloadWriter {
     waitFor(count, timeout, this.@schemaRegistryUsages)
   }
 
-  private static void waitFor(int count, long timeout, Collection collection) {
+  StatsGroup waitForGroup(Closure<Boolean> predicate, long timeout = TimeUnit.SECONDS.toMillis(3)) {
+    long deadline = System.currentTimeMillis() + timeout
+    while (System.currentTimeMillis() < deadline) {
+      synchronized (this) {
+        StatsGroup group = this.@groups.find(predicate)
+        if (group != null) {
+          return group
+        }
+      }
+      Thread.sleep(100)
+    }
+
+    assert false: "Expected a matching stats group within ${timeout}ms"
+  }
+
+  private void waitFor(int count, long timeout, Collection collection) {
     long deadline = System.currentTimeMillis() + timeout
     while (System.currentTimeMillis() < deadline) {
       synchronized (this) {
diff --git a/dd-java-agent/instrumentation-testing/src/test/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriterTest.groovy b/dd-java-agent/instrumentation-testing/src/test/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriterTest.groovy
new file mode 100644
index 00000000000..e3706feb98f
--- /dev/null
+++ b/dd-java-agent/instrumentation-testing/src/test/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriterTest.groovy
@@ -0,0 +1,58 @@
+package datadog.trace.agent.test.datastreams
+
+import static datadog.trace.api.datastreams.DataStreamsTags.Direction.INBOUND
+import static java.util.concurrent.TimeUnit.SECONDS
+
+import datadog.trace.api.datastreams.DataStreamsTags
+import datadog.trace.api.datastreams.StatsPoint
+import datadog.trace.core.datastreams.StatsBucket
+import datadog.trace.core.datastreams.StatsGroup
+import spock.lang.Specification
+
+class RecordingDatastreamsPayloadWriterTest extends Specification {
+  private static Thread startWorker(RecordingDatastreamsPayloadWriter writer, long delay) {
+    return Thread.startDaemon {
+      writer.writePayload([bucket(1L, 123L, DataStreamsTags.create("test1", INBOUND))], null)
+      Thread.sleep(delay)
+      writer.writePayload([bucket(2L, 0L, DataStreamsTags.create("test2", INBOUND))], null)
+    }
+  }
+
+
+  def "waitForGroup waits for a matching group"() {
+    setup:
+    def writer = new RecordingDatastreamsPayloadWriter()
+    def worker = startWorker(writer, 50L)
+
+    when:
+    StatsGroup group = writer.waitForGroup({ it.parentHash == 0 }, SECONDS.toMillis(1))
+
+    then:
+    group.parentHash == 0
+
+    cleanup:
+    worker.join(SECONDS.toMillis(2))
+  }
+
+  def "waitForGroup fails by timeout"() {
+    setup:
+    def writer = new RecordingDatastreamsPayloadWriter()
+    def worker = startWorker(writer, SECONDS.toMillis(3))
+
+    when:
+    writer.waitForGroup({ it.parentHash == 0 }, SECONDS.toMillis(1))
+
+    then:
+    AssertionError error = thrown()
+    error.message.contains("Expected a matching stats group within 1000ms")
+
+    cleanup:
+    worker.join(SECONDS.toMillis(5))
+  }
+
+  private static StatsBucket bucket(long hash, long parentHash, DataStreamsTags tags) {
+    StatsBucket bucket = new StatsBucket(0, SECONDS.toNanos(1L))
+    bucket.addPoint(new StatsPoint(tags, hash, parentHash, hash, 0, 1, 1, 0, null))
+    return bucket
+  }
+}