[Flaky Test] TTestKafkaPublish/run_test(1):_publish_single_event_with_topic_from_type

[khushijain21]

Flaky Test

• Test Name: TestKafkaPublish/run_test(1):_publish_single_event_with_topic_from_type
• Link: https://buildkite.com/elastic/beats-libbeat/builds/30091#019e5ead-c924-421c-a7d5-48fd0c0f9c2e

Stack Trace

    logger.go:146: 2026-05-25T10:38:39.375Z	INFO	kafka	kafka message: Initializing new client
    logger.go:146: 2026-05-25T10:38:39.375Z	INFO	kafka	kafka message: Successfully initialized new client
    logger.go:146: 2026-05-25T10:38:39.375Z	INFO	kafka	client/metadata fetching metadata for [log-type-6055449951074597455] from broker localhost:9094
    logger.go:146: 2026-05-25T10:38:39.375Z	INFO	kafka	Connected to broker at localhost:9094 (unregistered)
    logger.go:146: 2026-05-25T10:38:39.615Z	INFO	kafka	client/brokers registered new broker #0 at localhost:9094
    logger.go:146: 2026-05-25T10:38:39.615Z	INFO	kafka	kafka message: client/metadata found some partitions to be leaderless
    logger.go:146: 2026-05-25T10:38:39.866Z	INFO	kafka	client/metadata retrying after 250ms... (2 attempts remaining)
    logger.go:146: 2026-05-25T10:38:39.866Z	INFO	kafka	client/metadata fetching metadata for [log-type-6055449951074597455] from broker localhost:9094
    logger.go:146: 2026-05-25T10:38:39.866Z	INFO	kafka	Connected to broker at localhost:9094 (registered as #0)
    logger.go:146: 2026-05-25T10:38:39.872Z	INFO	kafka	kafka message: client/metadata found some partitions to be leaderless
    logger.go:146: 2026-05-25T10:38:40.122Z	INFO	kafka	client/metadata retrying after 250ms... (1 attempts remaining)
    logger.go:146: 2026-05-25T10:38:40.122Z	INFO	kafka	client/metadata fetching metadata for [log-type-6055449951074597455] from broker localhost:9094
    logger.go:146: 2026-05-25T10:38:40.125Z	INFO	kafka	kafka message: client/metadata found some partitions to be leaderless
    logger.go:146: 2026-05-25T10:38:40.375Z	INFO	kafka	client/metadata retrying after 250ms... (0 attempts remaining)
    logger.go:146: 2026-05-25T10:38:40.376Z	INFO	kafka	client/metadata fetching metadata for [log-type-6055449951074597455] from broker localhost:9094
    logger.go:146: 2026-05-25T10:38:40.379Z	INFO	kafka	kafka message: client/metadata found some partitions to be leaderless
    logger.go:146: 2026-05-25T10:38:40.379Z	DEBUG	kafka	finished kafka batch
    logger.go:146: 2026-05-25T10:38:40.379Z	ERROR	kafka	Kafka publish failed with: kafka server: In the middle of a leadership election, there is currently no leader for this partition and hence it is unavailable for writes
    kafka_integration_test.go:314:
        	Error Trace:	/opt/buildkite-agent/builds/bk-agent-prod-gcp-1779704923432404915/elastic/beats-libbeat/libbeat/outputs/kafka/kafka_integration_test.go:655
        	            				/opt/buildkite-agent/builds/bk-agent-prod-gcp-1779704923432404915/elastic/beats-libbeat/libbeat/outputs/kafka/kafka_integration_test.go:314
        	Error:      	Not equal:
        	            	expected: 0x0
        	            	actual  : 0x4
        	Test:       	TestKafkaPublish/run_test(1):_publish_single_event_with_topic_from_type
        	Messages:   	expected batch to be ACKed, got 4

[botelastic]

This issue doesn't have a Team:<team> label.

[github-actions]

tl;dr: the strongest actionable lead is a loop-control bug still present in testReadFromKafkaTopic that can leak/spin partition reader goroutines and destabilize TestKafkaPublish, especially around run test(1); I recommend fixing that helper first, then re-running the flaky scenario under stress.

Recommendation

Patch libbeat/outputs/kafka/kafka_integration_test.go to make partition reader goroutines exit correctly (instead of break-ing only the select) and to avoid advancing shared offsets before the message is safely handed off. This is the most concrete, code-backed issue currently visible in the flaky path.

Findings

1. The current helper used by TestKafkaPublish has a control-flow bug in the reader goroutine:
   • libbeat/outputs/kafka/kafka_integration_test.go:617-627
   • It uses break inside select for both !ok and <-done, which exits only the select, not the outer for loop.
2. Shared state and ordering in the same helper can amplify flakiness across subtests:
   • Global offsets map: libbeat/outputs/kafka/kafka_integration_test.go:586
   • Offset is advanced before msgs <- msg: libbeat/outputs/kafka/kafka_integration_test.go:623-624
3. This flaky pattern is historically associated with run_test(1): publish single event with topic from type:
   • Prior issue: #44032 (panic: test timed out after 10m0s) with the same subtest name in stack trace.
4. Related prior fix (#50798) addressed part of this area (ACK gating + labeled break in the outer read loop), but the inner goroutine loop still contains bare break in current main (libbeat/outputs/kafka/kafka_integration_test.go:617-627).

Verification

I ran the target subtest in this workspace:

$ GOTOOLCHAIN=auto go test -v -count=1 -run 'TestKafkaPublish/run test\(1\): publish single event with topic from type' -tags integration ./libbeat/outputs/kafka
=== RUN   TestKafkaPublish
=== RUN   TestKafkaPublish/run_test(1):_publish_single_event_with_topic_from_type
...
Kafka publish failed with: kafka: client has run out of available brokers to talk to: dial tcp [::1]:9094: connect: connection refused
kafka_integration_test.go:655: expected batch to be ACKed, got 4
--- FAIL: TestKafkaPublish/run_test(1):_publish_single_event_with_topic_from_type

This environment does not have a reachable Kafka broker, so I could not reproduce CI flakiness end-to-end here; however, the code-level loop bug above is directly verifiable.

Detailed Action Plan

1. In libbeat/outputs/kafka/kafka_integration_test.go:617-627, replace bare break paths with termination of the goroutine loop (return or labeled break from the outer loop).
2. In the same goroutine, move offset advancement so it happens after successful handoff, or otherwise guarantee message delivery/ordering relative to offset update (:623-624).
3. Keep the existing ACK gate (requiretBatchesACKed, :649-657) and run targeted stress on the flaky subtest with Kafka available:
   • go test -v -race -tags integration -run 'TestKafkaPublish/run_test\(1\):_publish_single_event_with_topic_from_type' -count=50 ./libbeat/outputs/kafka
   • optionally script/stresstest.sh --tags integration ./libbeat/outputs/kafka '^TestKafkaPublish/run test\(1\): publish single event with topic from type$' -p 8
4. If flakes remain after loop fix, add a bounded retry around topic partition discovery in testReadFromKafkaTopic (libbeat/outputs/kafka/kafka_integration_test.go:599-603) to absorb transient metadata visibility lag for newly selected dynamic topics.

Related Items

Type	Link	Relevance
Issue	#44032	Historical flaky report with same subtest (run_test(1)) timing out
PR	#50798	Prior partial fix in same test (ACK check + outer read-loop timeout break)
File	libbeat/outputs/kafka/kafka_integration_test.go:617-627	Inner goroutine loop currently uses bare break in select
File	libbeat/outputs/kafka/kafka_integration_test.go:586,607,623-624	Shared offsets + update-before-handoff behavior
File	libbeat/outputs/kafka/client.go:236-248	Dynamic topic selection path used by "topic": "%{[type]}"

Note

🔒 Integrity filter blocked 15 items

The following items were blocked because they don't meet the GitHub integrity level.

• [Flaky Test] TTestKafkaPublish/run_test(1):_publish_single_event_with_topic_from_type #50899 issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50899 search_issues: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50899 issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50897 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50444 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #45327 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #31234 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #31284 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #31009 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #30922 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50838 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50837 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50836 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #30860 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #32493 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".

To allow these resources, lower min-integrity in your GitHub frontmatter:

tools:
  github:
    min-integrity: approved  # merged | approved | unapproved | none

---

What is this? | From workflow: Issue Triage

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.