From 5623407b74e964797f396a85cb51d7c90792ce3f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Miko=C5=82aj=20=C5=9Awi=C4=85tek?= <mail@mikolajswiatek.com>
Date: Wed, 15 Jul 2026 00:24:47 +0200
Subject: [PATCH] Set shutdown timeout for beat receivers to 5s by default
 (#51940)

Beat receivers need a longer timeout than the default 1s, as it needs to
exceed the batcher flush timeout. The current code gives the impression
of this already happening, but in reality, if the beater handles
pipeline shutdown on its own, the default timeout wins.

(cherry picked from commit 8f4bbab772a2263185d6d17965700603cbff0b34)
---
 x-pack/filebeat/fbreceiver/receiver_test.go | 93 +++++++++++++++++++++
 x-pack/libbeat/cmd/instance/beat.go         | 12 +++
 2 files changed, 105 insertions(+)

diff --git a/x-pack/filebeat/fbreceiver/receiver_test.go b/x-pack/filebeat/fbreceiver/receiver_test.go
index 240ac59863b2..e456852afe4b 100644
--- a/x-pack/filebeat/fbreceiver/receiver_test.go
+++ b/x-pack/filebeat/fbreceiver/receiver_test.go
@@ -20,12 +20,16 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/gofrs/uuid/v5"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -701,6 +705,95 @@ func TestConsumeContract(t *testing.T) {
 	})
 }
 
+// TestShutdownDrainTimeout is a regression test for the filebeat receiver
+// shutdown drain duration.
+//
+// The slow consumer blocks until the drain context is cancelled, so the
+// elapsed shutdown time equals the drain timeout. We only assert a lower bound
+// (>= receiverPublisherCloseTimeout - 500ms) rather than an upper bound, so
+// the test never flakes due to CI load while still catching any regression that
+// reduces the drain timeout below the expected value.
+func TestShutdownDrainTimeout(t *testing.T) {
+	defer oteltest.VerifyNoLeaks(t)
+
+	tmpDir := t.TempDir()
+	logFile := filepath.Join(tmpDir, "input.log")
+	for range 20 {
+		writeFile(t, logFile, `{"message": "drain timeout test"}`)
+	}
+
+	consumerCalledCh := make(chan struct{})
+	var consumerCalledOnce sync.Once
+	slowConsumer, err := consumer.NewLogs(func(ctx context.Context, _ plog.Logs) error {
+		consumerCalledOnce.Do(func() { close(consumerCalledCh) })
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(30 * time.Second):
+			return nil
+		}
+	})
+	require.NoError(t, err)
+
+	cfg := &Config{
+		Beatconfig: map[string]any{
+			"queue.mem.flush.timeout": "0s",
+			"filebeat": map[string]any{
+				"inputs": []map[string]any{
+					{
+						"type":                 "filestream",
+						"enabled":              true,
+						"id":                   "drain-timeout-test",
+						"paths":                []string{logFile},
+						"file_identity.native": map[string]any{},
+						"prospector": map[string]any{
+							"scanner": map[string]any{
+								"check_interval":      "50ms",
+								"fingerprint.enabled": false,
+							},
+						},
+					},
+				},
+			},
+			"path.home": tmpDir,
+			"path.logs": tmpDir,
+			"logging":   map[string]any{"level": "info"},
+		},
+	}
+
+	factory := NewFactoryWithSettings(Settings{Home: tmpDir})
+	var settings receiver.Settings
+	settings.ID = component.NewIDWithName(factory.Type(), "r1")
+	settings.Logger = zap.NewNop()
+
+	r, err := factory.CreateLogs(t.Context(), settings, cfg, slowConsumer)
+	require.NoError(t, err)
+	require.NoError(t, r.Start(t.Context(), componenttest.NewNopHost()))
+
+	// Guard: wait until events are in flight before triggering shutdown.
+	// Without this, shutdown starts before any batches exist and the drain
+	// completes trivially fast, making the test a no-op.
+	select {
+	case <-consumerCalledCh:
+	case <-time.After(30 * time.Second):
+		t.Fatal("ConsumeLogs was never called; events never reached the pipeline")
+	}
+
+	start := time.Now()
+	require.NoError(t, r.Shutdown(t.Context()))
+	elapsed := time.Since(start)
+
+	// Only a lower bound: CI load can make wall-clock time arbitrarily large,
+	// so an upper bound would cause spurious failures on loaded runners.
+	const (
+		receiverPublisherCloseTimeout = 5 * time.Second
+		minExpected                   = receiverPublisherCloseTimeout - 500*time.Millisecond
+	)
+	assert.GreaterOrEqual(t, elapsed, minExpected,
+		"shutdown completed in %v, expected >= %v; was shutdown_timeout injection removed?",
+		elapsed, minExpected)
+}
+
 func TestReceiverHook(t *testing.T) {
 	cfg := Config{
 		Beatconfig: map[string]any{
diff --git a/x-pack/libbeat/cmd/instance/beat.go b/x-pack/libbeat/cmd/instance/beat.go
index 1ad7d1e32fa5..aa1b5a8f2ffc 100644
--- a/x-pack/libbeat/cmd/instance/beat.go
+++ b/x-pack/libbeat/cmd/instance/beat.go
@@ -96,6 +96,18 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 		"otelconsumer": map[string]any{},
 	}
 
+	// Set the default shutdown timeout to 5s. The beat default is 1s, which can be too short for the otel pipeline.
+	switch beatSection := receiverConfig[b.Info.Beat].(type) {
+	case map[string]any:
+		if _, alreadySet := beatSection["shutdown_timeout"]; !alreadySet {
+			beatSection["shutdown_timeout"] = receiverPublisherCloseTimeout.String()
+		}
+	case nil:
+		receiverConfig[b.Info.Beat] = map[string]any{
+			"shutdown_timeout": receiverPublisherCloseTimeout.String(),
+		}
+	}
+
 	tmp, err := ucfg.NewFrom(receiverConfig, cfOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("error converting receiver config to ucfg: %w", err)