Add TestBatchSpanProcessorConcurrentSafe and fix deadlock

sdk/trace/batch_span_processor.go

@@ -195,6 +195,9 @@ func (bsp *batchSpanProcessor) ForceFlush(ctx context.Context) error {
 		flushCh := make(chan struct{})
 		if bsp.enqueueBlockOnQueueFull(ctx, forceFlushSpan{flushed: flushCh}) {
 			select {
+			case <-bsp.stopCh:
+				// The batchSpanProcessor is Shutdown.
+				return nil
 			case <-flushCh:
 				// Processed any items in queue prior to ForceFlush being called
 			case <-ctx.Done():

sdk/trace/batch_span_processor_test.go

@@ -594,6 +594,49 @@ func TestBatchSpanProcessorForceFlushQueuedSpans(t *testing.T) {
 	}
 }
 
+func TestBatchSpanProcessorConcurrentSafe(t *testing.T) {
+	ctx := context.Background()
+	var bp testBatchExporter
+	bsp := sdktrace.NewBatchSpanProcessor(&bp)
+	tp := basicTracerProvider(t)
+	tp.RegisterSpanProcessor(bsp)
+	tr := tp.Tracer(t.Name())
+
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		generateSpan(t, tr, testOption{genNumSpans: 1})
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_ = bsp.ForceFlush(ctx)
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_ = bsp.Shutdown(ctx)
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_ = tp.ForceFlush(ctx)
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_ = tp.Shutdown(ctx)
+	}()
+
+	wg.Wait()
+}
+
 func BenchmarkSpanProcessor(b *testing.B) {
 	tp := sdktrace.NewTracerProvider(
 		sdktrace.WithBatcher(