filebeat: stabilize kafka input integration topic readiness (#49228)

github-actions[bot] · Copilot · belimawr · web-flow · commit f6a7f39def09 · 2026-03-04T09:54:06.000-05:00
Use a shared helper that creates and waits for a test topic to become writable before producing events. Apply it across kafka integration tests, including TestTest, to reduce leader-election race flakes.

GenAI-Assisted: Yes
Human-Reviewed: Yes
Tool: GitHub Copilot CLI, Model: GPT-5.3 Codex

---------

Co-authored-by: github-actions[bot] &lt;41898282+github-actions[bot]@users.noreply.github.com&gt;
Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;
Co-authored-by: Tiago Queiroz &lt;tiago.queiroz@elastic.co&gt;
diff --git a/filebeat/input/kafka/kafka_integration_test.go b/filebeat/input/kafka/kafka_integration_test.go
@@ -67,7 +67,7 @@ func recordHeader(key, value string) sarama.RecordHeader {
 }
 
 func TestInput(t *testing.T) {
-	testTopic := createTestTopicName()
+	testTopic := createReadyTestTopic(t)
 	groupID := "filebeat"
 
 	// Send test messages to the topic for the input to read.
@@ -155,7 +155,7 @@ func TestInput(t *testing.T) {
 }
 
 func TestInputWithMultipleEvents(t *testing.T) {
-	testTopic := createTestTopicName()
+	testTopic := createReadyTestTopic(t)
 
 	// Send test messages to the topic for the input to read.
 	message := testMessage{
@@ -211,7 +211,7 @@ func TestInputWithMultipleEvents(t *testing.T) {
 }
 
 func TestInputWithJsonPayload(t *testing.T) {
-	testTopic := createTestTopicName()
+	testTopic := createReadyTestTopic(t)
 
 	// Send test message to the topic for the input to read.
 	message := testMessage{
@@ -273,7 +273,7 @@ func TestInputWithJsonPayload(t *testing.T) {
 }
 
 func TestInputWithJsonPayloadAndMultipleEvents(t *testing.T) {
-	testTopic := createTestTopicName()
+	testTopic := createReadyTestTopic(t)
 
 	// Send test messages to the topic for the input to read.
 	message := testMessage{
@@ -360,13 +360,9 @@ func TestSASLAuthentication(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			testTopic := createTestTopicName()
+			testTopic := createReadyTestTopic(t)
 			groupID := "filebeat"
 
-			// Topic auto-creation is asynchronous; wait for leaders to avoid transient
-			// "no leader for this partition" write failures in CI.
-			ensureKafkaTopicReadyForWrites(t, testTopic)
-
 			// Send test messages to the topic for the input to read.
 			messages := []testMessage{
 				{message: "testing"},
@@ -452,7 +448,7 @@ func TestSASLAuthentication(t *testing.T) {
 }
 
 func TestTest(t *testing.T) {
-	testTopic := createTestTopicName()
+	testTopic := createReadyTestTopic(t)
 
 	// Send test messages to the topic for the input to read.
 	message := testMessage{
@@ -483,8 +479,13 @@ func TestTest(t *testing.T) {
 	}
 }
 
-func createTestTopicName() string {
+func createReadyTestTopic(t *testing.T) string {
+	t.Helper()
+
 	testTopic := fmt.Sprintf("Filebeat-TestInput-%d", rand.Int())
+	// Topic auto-creation is asynchronous; explicitly wait for leaders to avoid
+	// transient "no leader for this partition" write failures in CI.
+	ensureKafkaTopicReadyForWrites(t, testTopic)
 	return testTopic
 }
 
