trace: fix span leak in batchSpanProcessor

sdk/trace/batch_span_processor.go

@@ -280,6 +280,7 @@ func (bsp *batchSpanProcessor) exportSpans(ctx context.Context) error {
 		//
 		// It is up to the exporter to implement any type of retry logic if a batch is failing
 		// to be exported, since it is specific to the protocol and backend being sent to.
+		clear(bsp.batch)
 		bsp.batch = bsp.batch[:0]
 
 		if err != nil {

sdk/trace/batch_span_processor_test.go

@@ -9,9 +9,11 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"reflect"
 	"sync"
 	"testing"
 	"time"
+	"unsafe"
 
 	ottest "go.opentelemetry.io/otel/sdk/internal/internaltest"
 
@@ -627,6 +629,34 @@ func TestBatchSpanProcessorConcurrentSafe(t *testing.T) {
 	wg.Wait()
 }
 
+func TestBatchSpanLeak(t *testing.T) {
+	ctx := context.Background()
+
+	getBatch := func(bsp sdktrace.SpanProcessor) []sdktrace.ReadOnlySpan {
+		batchField := reflect.ValueOf(bsp).Elem().FieldByName("batch")
+		require.True(t, batchField.IsValid())
+		return unsafe.Slice((*sdktrace.ReadOnlySpan)(batchField.UnsafePointer()), batchField.Cap())[:batchField.Len()]
+	}
+	bsp := sdktrace.NewBatchSpanProcessor(tracetest.NewInMemoryExporter(), []sdktrace.BatchSpanProcessorOption{}...)
+	tp := sdktrace.NewTracerProvider(
+		sdktrace.WithSpanProcessor(bsp),
+	)
+
+	_, span := tp.Tracer("tracer").Start(ctx, "leaked_span")
+	span.End()
+
+	err := tp.ForceFlush(ctx)
+	assert.NoError(t, err)
+
+	batch := getBatch(bsp)
+	assert.Equal(t, 0, len(batch))
+	batch = batch[:cap(batch)]
+
+	for _, span := range batch {
+		require.Nil(t, span)
+	}
+}
+
 func BenchmarkSpanProcessor(b *testing.B) {
 	tp := sdktrace.NewTracerProvider(
 		sdktrace.WithBatcher(