Modify ForceFlush to abort after timeout/cancellation

Author: humivo

CHANGELOG.md

@@ -13,6 +13,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Jaeger exporter was updated to use thrift v0.14.1. (#1712)
 - Migrate from using internally built and maintained version of the OTLP to the one hosted at `go.opentelemetry.io/proto/otlp`. (#1713)
 - Migrate from using `github.com/gogo/protobuf` to `google.golang.org/protobuf` to match `go.opentelemetry.io/proto/otlp`. (#1713)
+- Modify `BatchSpanProcessor.ForceFlush` to abort after timeout/cancellation (#1618)
 
 ### Removed
 

sdk/trace/batch_span_processor.go

@@ -148,7 +148,23 @@ func (bsp *batchSpanProcessor) Shutdown(ctx context.Context) error {
 
 // ForceFlush exports all ended spans that have not yet been exported.
 func (bsp *batchSpanProcessor) ForceFlush(ctx context.Context) error {
-	return bsp.exportSpans(ctx)
+	var err error
+	if bsp.e != nil {
+		wait := make(chan struct{})
+		go func() {
+			if err := bsp.exportSpans(ctx); err != nil {
+				otel.Handle(err)
+			}
+			close(wait)
+		}()
+		// Wait until the wait group is done or the context is cancelled/timed out
+		select {
+		case <-wait:
+		case <-ctx.Done():
+			err = ctx.Err()
+		}
+	}
+	return err
 }
 
 func WithMaxQueueSize(size int) BatchSpanProcessorOption {

sdk/trace/batch_span_processor_test.go

@@ -257,3 +257,33 @@ func TestBatchSpanProcessorShutdown(t *testing.T) {
 	}
 	assert.Equal(t, 1, bp.shutdownCount)
 }
+
+func TestBatchSpanProcessorForceFlushSucceeds(t *testing.T) {
+	var bp testBatchExporter
+	bsp := sdktrace.NewBatchSpanProcessor(&bp)
+
+	err := bsp.ForceFlush(context.Background())
+	assert.Equal(t, nil, err)
+}
+
+func TestBatchSpanProcessorForceFlushTimeout(t *testing.T) {
+	var bp testBatchExporter
+	bsp := sdktrace.NewBatchSpanProcessor(&bp)
+	// Add timeout to context to test deadline
+	ctx, cancel := context.WithTimeout(context.Background(), 0)
+	defer cancel()
+
+	err := bsp.ForceFlush(ctx)
+	assert.Equal(t, context.DeadlineExceeded, err)
+}
+
+func TestBatchSpanProcessorForceFlushCancellation(t *testing.T) {
+	var bp testBatchExporter
+	bsp := sdktrace.NewBatchSpanProcessor(&bp)
+	ctx, cancel := context.WithCancel(context.Background())
+	// Cancel the context
+	cancel()
+
+	err := bsp.ForceFlush(ctx)
+	assert.Equal(t, context.Canceled, err)
+}