Fix batch processor shutdown deadlock & goroutine leak under load

[WHOIM1205]

Summary

This PR fixes a critical shutdown deadlock and goroutine leak in the core batch processor caused by blocking channel sends that ignore context cancellation and shutdown signals.

Under high load, this bug can cause the OpenTelemetry Collector to hang indefinitely during shutdown, resulting in SIGKILL by Kubernetes and 100% loss of buffered telemetry.

---

Problem Description

The batch processor’s consume() path performs an unconditional blocking send to an internal buffered channel.

Key issues:

• consume() ignores context.Context
• Channel buffer is bounded
• Blocked senders are never released during shutdown

During shutdown:

1. Receivers shut down first (topological order)
2. In-flight requests block in consume()
3. Batch processor shutdown never completes
4. Collector hangs until force-killed

This leads to deadlock, goroutine leaks, and silent telemetry loss.

---

Affected Code

• processor/batchprocessor/batch_processor.go
• singleShardBatcher.consume
• multiShardBatcher.consume

---

Root Cause

The batch processor assumes:

• Channel sends will not block for long
• Draining buffers during shutdown is sufficient

These assumptions break under load:

• Producers can block indefinitely
• Shutdown does not signal or unblock senders
• Context cancellation is ignored

---

Fix

Make consume() context- and shutdown-aware:

• Replace unconditional channel send with select
• Respect:
  • ctx.Done() (request cancellation / deadlines)
  • shutdownC (processor shutdown signal)
• Return an error instead of blocking forever

This preserves existing behavior while preventing deadlocks and leaks.

---

Tests Added

Shutdown Safety

• Shutdown while sends are blocked
• Single-shard and multi-shard batchers
• Traces, Metrics, Logs

Context Cancellation

• Consume* respects request timeouts
• Returns context.DeadlineExceeded instead of blocking

Regression Coverage

• Normal batch behavior remains unchanged

---

Reproduction (Before Fix)

1. Start Collector with:
   • otlp receiver
   • batch processor
   • Slow exporter
2. Send high-volume telemetry continuously
3. Trigger shutdown (SIGTERM / pod eviction)
4. Observe:
   • Shutdown hangs
   • Collector is SIGKILLed
   • Buffered telemetry is lost

---

Impact

Before

• Shutdown deadlock
• Goroutine leaks
• SIGKILL during rolling updates
• Silent loss of buffered telemetry

After

• Clean and fast shutdown
• No blocked goroutines
• Proper cancellation handling
• Telemetry integrity preserved

---

Risk Assessment

• Low risk
• Minimal, localized change
• Comprehensive test coverage
• No steady-state behavior change

[iblancasa]

Avoid time.Sleep calls, please.

[iblancasa]

It is not needed to comment all the lines

[iblancasa]

Depending on the component sending the data, I think this can be retried if it is not marked as permanent.

[jmacd]

See also #14473

See also #13583

I do not think we should maintain this component.