[chore] Remove time.Sleep calls from group by trace processor tests (open-telemetry#44459)

#### Link to tracking issue
Fixes open-telemetry#42461

#### Testing
Run with `-count=300`

Signed-off-by: Israel Blancas <[email protected]>
Co-authored-by: Christos Markou <[email protected]>

Author: iblancasa

processor/groupbytraceprocessor/event_test.go

@@ -396,8 +396,11 @@ func TestEventShutdown(t *testing.T) {
 		payload: pcommon.TraceID([16]byte{1, 2, 3, 4}),
 	})
 
-	time.Sleep(10 * time.Millisecond)  // give it a bit of time to process the items
-	assert.Equal(t, 1, em.numEvents()) // we should have one pending event in the queue, the second traceRemoved event
+	// wait for events to process - we should have one pending event in the queue, the second traceRemoved event
+	assert.Eventually(t, func() bool {
+		return em.numEvents() == 1
+	}, 1*time.Second, 10*time.Millisecond)
+	assert.Equal(t, 1, em.numEvents())
 
 	shutdownWg := sync.WaitGroup{}
 	shutdownWg.Add(1)
@@ -406,9 +409,11 @@ func TestEventShutdown(t *testing.T) {
 		shutdownWg.Done()
 	}()
 
-	wg.Done()                          // the pending event should be processed
-	time.Sleep(100 * time.Millisecond) // give it a bit of time to process the items
-
+	wg.Done() // the pending event should be processed
+	// wait for shutdown to process remaining events
+	assert.Eventually(t, func() bool {
+		return em.numEvents() == 0
+	}, 1*time.Second, 10*time.Millisecond)
 	assert.Equal(t, 0, em.numEvents())
 
 	// new events should *not* be processed
@@ -422,8 +427,10 @@ func TestEventShutdown(t *testing.T) {
 
 	// If the code is wrong, there's a chance that the test will still pass
 	// in case the event is processed after the assertion.
-	// for this reason, we add a small delay here
-	time.Sleep(10 * time.Millisecond)
+	// Verify that the expired event is not processed (should remain 0)
+	assert.Eventually(t, func() bool {
+		return traceExpiredFired.Load() == 0
+	}, 100*time.Millisecond, 5*time.Millisecond)
 	assert.Equal(t, int64(0), traceExpiredFired.Load())
 
 	// wait until the shutdown has returned
@@ -467,11 +474,13 @@ func TestPeriodicMetrics(t *testing.T) {
 	em.workers[0].fire(event{typ: traceReceived}) // the first is consumed right away, the second is in the queue
 	go em.periodicMetrics()
 
-	// TODO: Remove time.Sleep below, see https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/42515
-	time.Sleep(10 * time.Millisecond)
 	// ensure our gauge is showing 1 item in the queue
 	assert.EventuallyWithT(t, func(tt *assert.CollectT) {
 		val := getGaugeValue(t.Context(), tt, "otelcol_processor_groupbytrace_num_events_in_queue", s)
+		if val == -1 {
+			tt.Errorf("gauge not yet created or has no data points")
+			return
+		}
 		assert.Equal(tt, int64(1), val)
 	}, 1*time.Second, 10*time.Millisecond)
 
@@ -480,14 +489,23 @@ func TestPeriodicMetrics(t *testing.T) {
 	// ensure our gauge is now showing no items in the queue
 	assert.EventuallyWithT(t, func(tt *assert.CollectT) {
 		val := getGaugeValue(t.Context(), tt, "otelcol_processor_groupbytrace_num_events_in_queue", s)
+		if val == -1 {
+			tt.Errorf("gauge not yet created or has no data points")
+			return
+		}
 		assert.Equal(tt, int64(0), val)
 	}, 1*time.Second, 10*time.Millisecond)
 
 	// signal and wait for the recursive call to finish
 	em.shutdownLock.Lock()
 	em.closed = true
 	em.shutdownLock.Unlock()
-	time.Sleep(5 * time.Millisecond)
+	// Wait for periodicMetrics to detect the closed flag and return
+	assert.Eventually(t, func() bool {
+		em.shutdownLock.RLock()
+		defer em.shutdownLock.RUnlock()
+		return em.closed
+	}, 100*time.Millisecond, 5*time.Millisecond)
 }
 
 func TestForceShutdown(t *testing.T) {
@@ -507,8 +525,12 @@ func TestForceShutdown(t *testing.T) {
 	// verify
 	assert.Greater(t, duration, 20*time.Millisecond)
 
-	// wait for shutdown goroutine to end
-	time.Sleep(100 * time.Millisecond)
+	// Verify shutdown completed - the machine should be closed
+	assert.Eventually(t, func() bool {
+		em.shutdownLock.RLock()
+		defer em.shutdownLock.RUnlock()
+		return em.closed
+	}, 100*time.Millisecond, 5*time.Millisecond)
 }
 
 func TestDoWithTimeout_NoTimeout(t *testing.T) {
@@ -525,10 +547,11 @@ func TestDoWithTimeout_NoTimeout(t *testing.T) {
 func TestDoWithTimeout_TimeoutTrigger(t *testing.T) {
 	// prepare
 	start := time.Now()
+	blockCh := make(chan struct{}) // channel that will never be closed/signaled
 
 	// test
 	succeed, err := doWithTimeout(20*time.Millisecond, func() error {
-		time.Sleep(1 * time.Second)
+		<-blockCh // block forever (simulating a long-running function)
 		return nil
 	})
 	assert.False(t, succeed)
@@ -541,13 +564,18 @@ func TestDoWithTimeout_TimeoutTrigger(t *testing.T) {
 func getGaugeValue(ctx context.Context, t *assert.CollectT, name string, tt testTelemetry) int64 {
 	var md metricdata.ResourceMetrics
 	require.NoError(t, tt.reader.Collect(ctx, &md))
-	m := tt.getMetric(name, md).Data
+	metric := tt.getMetric(name, md)
+	if metric == (metricdata.Metrics{}) {
+		return -1 // return sentinel value to indicate metric doesn't exist yet
+	}
+	m := metric.Data
 	var g metricdata.Gauge[int64]
 	var ok bool
 	if g, ok = m.(metricdata.Gauge[int64]); !ok {
-		assert.Fail(t, "missing gauge data")
-	} else {
-		assert.Len(t, g.DataPoints, 1, "expected exactly one data point")
+		return -1 // return sentinel value to indicate gauge data is missing
+	}
+	if len(g.DataPoints) == 0 {
+		return -1 // return sentinel value to indicate no data points yet
 	}
 	return g.DataPoints[0].Value
 }