refactor: add `Context` parameter to `Backoff.Wait` method by andrzej-stencel · Pull Request #50428 · elastic/beats

andrzej-stencel · 2026-04-30T12:12:03Z

Proposed commit message

Changes the API of the Backoff interface, adding context.Context to its Wait method.

Previously, the Wait method used a context (or more specifically, a done channel) that was passed during creation of a Backoff instance, specifically in NewExpBackoff and NewEqualJitterBackoff functions.
This was not a correct design, as the proper context might not be available during creation of the Backoff instance. This led to confusion, see for example this discussion.

The changes are split into separate commits for each affected package. See the individual commit messages for details.

I had the changes planned and executed by Claude. I then went through each package one by one, scrutinizing each change and modifying where needed.

Checklist

My code follows the style guidelines of this project
~~[ ] I have commented my code, particularly in hard-to-understand areas~~
~~[ ] I have made corresponding changes to the documentation~~
~~[ ] I have made corresponding change to the default configuration files~~
[ ] I have added tests that prove my fix is effective or that my feature works. Where relevant, I have used the stresstest.sh script to run them under stress conditions and race detector to verify their stability.
~~[ ] I have added an entry in ./changelog/fragments using the changelog tool.~~

Disruptive User Impact

None expected, this is an internal API change.

The only theoretically functional change is in the Kafka input implementation, where the connectDelay now uses current available context, but previously it used context.Background().

How to test this PR locally

I ran unit tests for each modified package.

Removes `done` channel from backoff constructor functions. Uses the `Context` instance passed as parameter to `Backoff.Wait` method.

Moves retryBackoff instantiation to `MakeOtelConsumer`. Each `Backoff.Wait` method now uses the context from the current `Publish` call, removing confusion.

The call to `Backoff.Wait` gets the `f.readerCtx` which is functionally equivalent to the `canceler.Done()` channel used previously.

To create an instance of `context.Context` from `input.Context` (v1), I applied the same approach that exists for `input.Context` (v2) in `filebeat/input/kafka/input.go`.

Note that in `makeESClient`, the `connectDelay` previously used `context.Background`. This change uses the context passed to the `makeESClient` function.

infra-vault-gh-plugin-prod · 2026-04-30T12:12:08Z

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

github-actions · 2026-04-30T12:12:18Z

🤖 GitHub comments

Just comment with:

run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)
/test : Run the Buildkite pipeline.

coderabbitai · 2026-04-30T12:16:00Z

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Enterprise

Run ID: fb83ca47-541b-4ac0-beea-7f943ea2844a

📥 Commits

Reviewing files that changed from the base of the PR and between c42e1c4 and 1b75ee8.

📒 Files selected for processing (1)

heartbeat/beater/heartbeat.go

🚧 Files skipped from review as they are similar to previous changes (1)

heartbeat/beater/heartbeat.go

📝 Walkthrough

Walkthrough

The backoff API was changed to make waits cancelable via contexts: Backoff.Wait(ctx context.Context) bool. Factory functions (NewExpBackoff, NewEqualJitterBackoff) no longer take a done channel. WaitOnError now accepts a context.Context. Call sites across Filebeat, Heartbeat, libbeat, outputs (including Redis), MQTT, Kafka, journald, AWS S3, Azure Event Hub, otelconsumer, and tests were updated to create backoffs without a done channel and to pass contexts into Wait(...) calls.

🚥 Pre-merge checks | ✅ 2

✅ Passed checks (2 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

Create PR with unit tests

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

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

_{Review rate limit: 7/8 reviews remaining, refill in 7 minutes and 30 seconds.}

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

For the forbidigo errors forbidding usage of global logger, I just added nolint directives to minimize the amount of unrelated changes in this PR. In my view, the loggers issue should be fixed separately and globally (not only in the files touched by this PR).

github-actions · 2026-04-30T15:27:20Z

TL;DR

Both failed Buildkite jobs are failing on the same Filebeat integration test (TestDataAddedAfterCloseInactive) due to a timing-sensitive log assertion, not a failure in this PR’s changed code. The immediate fix is to relax or replace the "Harvester already running" log assertion with a state-based check.

Remediation

In filebeat/input/filestream/input_integration_test.go:1413, replace env.WaitLogsContains("Harvester already running", 2*time.Second) with a deterministic assertion based on behavior/state (for example: assert no duplicate harvester starts for the same source, or gate this assertion behind a stronger precondition that the old harvester is still active at write-event time).
If keeping the log assertion, increase the timing margin and add synchronization around the write-event/close-inactive boundary so the expected branch is guaranteed.
Re-run:
- cd filebeat && mage goIntegTest
- cd filebeat && mage goFIPSOnlyIntegTest

Investigation details

Root Cause

The two failing jobs report the identical failure:

filebeat/input/filestream/input_integration_test.go:1413 expects log line "Harvester already running" within 2*time.Second.
That log is emitted only in filebeat/input/filestream/internal/input-logfile/harvester.go:233 and filebeat/input/filestream/internal/input-logfile/harvester.go:271.
The test timing config (close.on_state_change.inactive: 1s, backoff.init/max: 3s) creates a race window around the inactive-close and subsequent write event (input_integration_test.go:1385-1389, 1407-1413). On slower/variant scheduling (including FIPS runs), the old harvester can be past that branch before the assertion window closes, so behavior can still be correct while this specific log line is absent.

This PR commit (1b75ee861f6dfbb1f11f75efc880e3a013965737) modifies heartbeat/beater/heartbeat.go and does not touch filebeat/input/filestream, which supports this being unrelated test flakiness.

Evidence

Build: https://buildkite.com/elastic/beats/builds/45255
Failed jobs:
- :ubuntu: Filebeat: Go Integration Tests
- :ubuntu: Filebeat: Go fips140=only Integration Tests
Key log excerpt (both jobs):

=== FAIL: filebeat/input/filestream TestDataAddedAfterCloseInactive (4.01s)

input_integration_test.go:1413: did not find 'Harvester already running' in the logs

Verification

Ran locally: cd filebeat && go test -tags integration -run TestDataAddedAfterCloseInactive -count=5 ./input/filestream
Result: passed locally (non-reproducible here), consistent with a timing-sensitive/flaky assertion.

Follow-up

If you want, I can narrow this further to a concrete replacement assertion that validates single-harvester semantics without depending on debug log timing.

Note

🔒 Integrity filter blocked 2 items

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

#50428 pull_request_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
refactor: add Context parameter to Backoff.Wait method #50428 pull_request_read: 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: PR Buildkite Detective

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

mauri870 · 2026-05-04T14:49:20Z

 			startedAt := time.Now()
-			assert.True(t, WaitOnError(b, errors.New("bad bad")))
-			assert.GreaterOrEqual(t, time.Now().Sub(startedAt), init)
+			assert.True(t, WaitOnError(context.Background(), b, errors.New("bad bad")))


Suggested change

assert.True(t, WaitOnError(context.Background(), b, errors.New("bad bad")))

assert.True(t, WaitOnError(t.Context(), b, errors.New("bad bad")))

mauri870 · 2026-05-04T14:50:04Z

-			b := f(c)
-			close(c)
-			assert.False(t, b.Wait())
+			ctx, cancel := context.WithCancel(context.Background())


Can we use t.Context()?

mauri870

Codewise looks good, pending the CI issues.

andrzej-stencel added 11 commits April 30, 2026 11:47

refactor!: add Context to Backoff.Wait

72e5258

refactor!: apply changed Backoff API in backoff package

090c767

Removes `done` channel from backoff constructor functions. Uses the `Context` instance passed as parameter to `Backoff.Wait` method.

refactor(otelconsumer): apply backoff API change

835226d

Moves retryBackoff instantiation to `MakeOtelConsumer`. Each `Backoff.Wait` method now uses the context from the current `Publish` call, removing confusion.

refactor(input/filestream): apply Backoff API change

5c2678f

The call to `Backoff.Wait` gets the `f.readerCtx` which is functionally equivalent to the `canceler.Done()` channel used previously.

refactor(input/journald): apply Backoff API change

854d57a

refactor(input/kafka): apply Backoff API change

c1a37d2

refactor(input/mqtt): apply Backoff API change

4f54ba0

To create an instance of `context.Context` from `input.Context` (v1), I applied the same approach that exists for `input.Context` (v2) in `filebeat/input/kafka/input.go`.

refactor(heartbeat): apply Backoff API change

4a542cd

Note that in `makeESClient`, the `connectDelay` previously used `context.Background`. This change uses the context passed to the `makeESClient` function.

refactor(input/awss3): apply Backoff API change

6f56c81

refactor(input/azureeventhub): apply Backoff API change

e47d686

refactor(output): apply Backoff API changes to backoffClient types

53a2e8c

andrzej-stencel requested review from a team as code owners April 30, 2026 12:12

andrzej-stencel added cleanup backport-skip Skip notification from the automated backport with mergify Team:Elastic-Agent-Data-Plane Label for the Agent Data Plane team labels Apr 30, 2026

andrzej-stencel requested review from faec and mauri870 April 30, 2026 12:12

andrzej-stencel added the skip-changelog label Apr 30, 2026

botelastic Bot added the needs_team Indicates that the issue/PR needs a Team:* label label Apr 30, 2026

botelastic Bot removed the needs_team Indicates that the issue/PR needs a Team:* label label Apr 30, 2026

mergify Bot assigned andrzej-stencel Apr 30, 2026

This comment has been minimized.

Sign in to view

andrzej-stencel added 2 commits April 30, 2026 14:51

fix lint errors

c42e1c4

For the forbidigo errors forbidding usage of global logger, I just added nolint directives to minimize the amount of unrelated changes in this PR. In my view, the loggers issue should be fixed separately and globally (not only in the files touched by this PR).

fix lint errors

1b75ee8

This comment has been minimized.

Sign in to view

github-actions Bot mentioned this pull request May 4, 2026

[agent-suggestions] Agent suggestions - 2026-05-04 #50476

Open

4 tasks

mauri870 reviewed May 4, 2026

View reviewed changes

mauri870 approved these changes May 4, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor: add `Context` parameter to `Backoff.Wait` method#50428

refactor: add `Context` parameter to `Backoff.Wait` method#50428
andrzej-stencel wants to merge 13 commits intoelastic:mainfrom
andrzej-stencel:add-context-to-backoff-wait

andrzej-stencel commented Apr 30, 2026

Uh oh!

infra-vault-gh-plugin-prod Bot commented Apr 30, 2026

Uh oh!

github-actions Bot commented Apr 30, 2026

Uh oh!

coderabbitai Bot commented Apr 30, 2026 •

edited

Loading

Walkthrough

Uh oh!

This comment has been minimized.

This comment has been minimized.

github-actions Bot commented Apr 30, 2026

Root Cause

Evidence

Verification

Follow-up

Uh oh!

mauri870 May 4, 2026

Uh oh!

mauri870 May 4, 2026

Uh oh!

mauri870 left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	assert.True(t, WaitOnError(context.Background(), b, errors.New("bad bad")))
	assert.True(t, WaitOnError(t.Context(), b, errors.New("bad bad")))

Conversation

andrzej-stencel commented Apr 30, 2026

Proposed commit message

Checklist

Disruptive User Impact

How to test this PR locally

Uh oh!

infra-vault-gh-plugin-prod Bot commented Apr 30, 2026

Uh oh!

github-actions Bot commented Apr 30, 2026

🤖 GitHub comments

Uh oh!

coderabbitai Bot commented Apr 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Walkthrough

Uh oh!

This comment has been minimized.

This comment has been minimized.

github-actions Bot commented Apr 30, 2026

TL;DR

Remediation

Root Cause

Evidence

Verification

Follow-up

Uh oh!

mauri870 May 4, 2026

Choose a reason for hiding this comment

Uh oh!

mauri870 May 4, 2026

Choose a reason for hiding this comment

Uh oh!

mauri870 left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

coderabbitai Bot commented Apr 30, 2026 •

edited

Loading