Allow flush to return if executor already shut down
Otherwise it could block shutdown if executor is stopped before disruptor.
MODIFY
dd-trace-ot/src/main/java/datadog/trace/common/writer/DDAgentWriter.java
dd-trace-ot/src/main/java/datadog/trace/common/writer/DDAgentWriter.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
@@ -317,7 +318,7 @@ public class DDAgentWriter implements Writer {
}
if (event.shouldFlush || payloadSize >= FLUSH_PAYLOAD_BYTES) {
- boolean early = (payloadSize >= FLUSH_PAYLOAD_BYTES);
+ final boolean early = (payloadSize >= FLUSH_PAYLOAD_BYTES);
reportTraces(early);
event.shouldFlush = false;
@@ -333,6 +334,12 @@ public class DDAgentWriter implements Writer {
return;
// scheduleFlush called in finally block.
}
+ if (scheduledWriterExecutor.isShutdown()) {
+ monitor.onFailedSend(
+ DDAgentWriter.this, traceCount.get(), payloadSize, DDApi.Response.failed(-1));
+ apiPhaser.arrive(); // Allow flush to return
+ return;
+ }
final List<byte[]> toSend = serializedTraces;
serializedTraces = new ArrayList<>(toSend.size());
// ^ Initialize with similar size to reduce arraycopy churn.
@@ -340,6 +347,7 @@ public class DDAgentWriter implements Writer {
final int representativeCount = traceCount.getAndSet(0);
final int sizeInBytes = payloadSize;
+ try {
monitor.onFlush(DDAgentWriter.this, early);
// Run the actual IO task on a different thread to avoid blocking the consumer.
@@ -366,7 +374,8 @@ public class DDAgentWriter implements Writer {
if (response.success()) {
log.debug("Successfully sent {} traces to the API", toSend.size());
- monitor.onSend(DDAgentWriter.this, representativeCount, sizeInBytes, response);
+ monitor.onSend(
+ DDAgentWriter.this, representativeCount, sizeInBytes, response);
} else {
log.debug(
"Failed to send {} traces (representing {}) of size {} bytes to the API",
@@ -394,6 +403,11 @@ public class DDAgentWriter implements Writer {
}
}
});
+ } catch (final RejectedExecutionException ex) {
+ monitor.onFailedSend(
+ DDAgentWriter.this, representativeCount, sizeInBytes, DDApi.Response.failed(ex));
+ apiPhaser.arrive(); // Allow flush to return
+ }
} finally {
payloadSize = 0;
scheduleFlush();
@@ -541,7 +555,7 @@ public class DDAgentWriter implements Writer {
};
}
- private static final String tag(final String tagPrefix, final String tagValue) {
+ private static String tag(final String tagPrefix, final String tagValue) {
return tagPrefix + ":" + tagValue;
}
@@ -626,6 +640,7 @@ public class DDAgentWriter implements Writer {
}
}
+ @Override
public String toString() {
if (hostInfo == null) {
return "StatsD";

MODIFY
dd-trace-ot/src/test/groovy/datadog/trace/api/writer/DDAgentWriterTest.groovy
dd-trace-ot/src/test/groovy/datadog/trace/api/writer/DDAgentWriterTest.groovy
@@ -198,6 +198,25 @@ class DDAgentWriterTest extends DDSpecification {
writer.traceCount.get() == 0
}
+ def "check shutdown if executor stopped first"() {
+ setup:
+ def writer = new DDAgentWriter(api)
+ writer.start()
+ writer.scheduledWriterExecutor.shutdown()
+
+ when:
+ writer.write([])
+ writer.flush()
+
+ then:
+ 1 * api.serializeTrace(_) >> { trace -> callRealMethod() }
+ 0 * _
+ writer.traceCount.get() == 1
+
+ cleanup:
+ writer.close()
+ }
+
def createMinimalTrace() {
def minimalContext = new DDSpanContext(
1G,

Allow flush to return if executor already shut down

Otherwise it could block shutdown if executor is stopped before disruptor.