more stable testing strategy

dd-trace-core/src/test/groovy/datadog/trace/api/writer/DDAgentWriterTest.groovy

@@ -20,6 +20,7 @@ import spock.lang.Retry
 import spock.lang.Timeout
 
 import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
 import java.util.concurrent.Phaser
 import java.util.concurrent.Semaphore
 import java.util.concurrent.TimeUnit
@@ -676,10 +677,13 @@ class DDAgentWriterTest extends DDSpecification {
 
   def "test backed up buffer"() {
     setup:
+    def minimalTrace = createMinimalTrace()
     AtomicInteger backedUp = new AtomicInteger(0)
     def serializer = Spy(new MsgPackStatefulSerializer())
     def monitor = Mock(Monitor)
     def api = Mock(DDAgentApi)
+    def latch = new CountDownLatch(1)
+    def arbitrator = Executors.newSingleThreadScheduledExecutor()
     def dispatcher = new DispatchingDisruptor(2,
       DDAgentWriter.toEventFactory(serializer),
       api,
@@ -688,16 +692,24 @@ class DDAgentWriterTest extends DDSpecification {
     dispatcher.start()
 
     when:
-    // provoke temporal congestion, review this if the test proves to be flakey
-    api.sendSerializedTraces(_) >> { Thread.sleep(1000) }
+    api.sendSerializedTraces(_) >> {
+      // wait to be unblocked, by which time the test aims to have created a backlog
+      latch.await()
+      DDAgentApi.Response.success(200)
+    }
     monitor.onBackedUpTraceBuffer() >> { backedUp.incrementAndGet() }
     long txnId1 = dispatcher.beginTransaction()
-    dispatcher.getTraceBuffer(txnId1)
+    serializer.reset(dispatcher.getTraceBuffer(txnId1))
+    serializer.serialize(minimalTrace)
     dispatcher.commit(txnId1)
-    // if the code below can't run within a second, the test will fail,
-    // but this behaviour is hard to provoke with latches
+    arbitrator.schedule({
+      latch.countDown()
+    }, 1000, TimeUnit.MILLISECONDS)
+    // if the code below can't run within a second (when the latch is counted down),
+    // the test will fail, but this behaviour is hard to provoke with latches
     long txnId2 = dispatcher.beginTransaction()
-    dispatcher.getTraceBuffer(txnId2)
+    serializer.reset(dispatcher.getTraceBuffer(txnId2))
+    serializer.serialize(minimalTrace)
     dispatcher.commit(txnId2)
     long txnId3 = dispatcher.beginTransaction()
 
@@ -707,6 +719,7 @@ class DDAgentWriterTest extends DDSpecification {
 
     cleanup:
     dispatcher.close()
+    arbitrator.shutdownNow()
   }
 
   static int calculateSize(List<DDSpan> trace) {