[fix] kafka input Test() panics when broker connection fails

[barkhayot]

• When brokers are unreachable, sarama.NewClient() fails and returns a nil client. The error was only logged without returning, so execution continued to client.Topics() on a nil client, causing a panic instead of a clean error [bug-hunter] Kafka input Test() panics when broker connection fails #49376
• Return the error from sarama.NewClient() immediately so Test() exits cleanly when broker connection fails
• Add suggested test from issue.

Closes #49376.

[github-actions]

🤖 GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @barkhayot? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: d0678fed-b640-4a87-8774-114447b806fb

📥 Commits

Reviewing files that changed from the base of the PR and between 12e4652 and 9890216.

📒 Files selected for processing (2)

• filebeat/input/kafka/input.go
• filebeat/input/kafka/input_test.go

---

📝 Walkthrough

Walkthrough

The changes modify the Kafka input error handling behavior. When client initialization fails in the Test method of kafkaInput, the function now returns the error immediately instead of logging it and continuing execution. This prevents subsequent operations from attempting to use an invalid client. A new test case TestTestNoPanicOnClientError was added to verify this behavior by attempting client initialization with invalid host/port configuration and asserting that an error is returned.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

• 🛠️ Update Documentation: Commit on current branch
• 🛠️ Update Documentation: Create PR

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[mauri870]

For instance, this is a panic in the input package but in a test-only function, the input itself is not affected. Either way we should prevent panics from happening.

Updated your PR description to include a closes comment.

[barkhayot]

For instance, this is a panic in the input package but in a test-only function, the input itself is not affected. Either way we should prevent panics from happening.

Updated your PR description to include a closes comment.

@mauri870 Thanks for comment! Will be opening a follow-up PR later on to address the same issue in the input itself as well.

[mauri870]

Suggested change

	// AssertNotStartedInputCanBeDone checks that the context of an input can be
	// done before starting the input, and it doesn't leak goroutines. This is
	// important to confirm that leaks don't happen with CheckConfig.
	func TestTestNoPanicOnClientError(t *testing.T) {
	// TestNoPanicOnClientError checks that the context of an input can be
	// done before starting the input, and it doesn't leak goroutines. This is
	// important to confirm that leaks don't happen with CheckConfig.
	func TestNoPanicOnClientError(t *testing.T) {