enhancement(kubernetes_logs source): add end-to-end acknowledgement support

[connoryy]

Summary

Adds end-to-end acknowledgement support to the kubernetes_logs source. When acknowledgements are enabled (via a downstream sink), file checkpoints only advance after downstream sinks confirm event delivery. This prevents data loss when the source crashes or restarts — unacknowledged events are re-read from the checkpoint position.

Based on initial work by @ganelo (Orri), cleaned up and rebased onto current master.

Motivation

The kubernetes_logs source currently returns can_acknowledge() -> false, meaning it cannot participate in Vector's end-to-end acknowledgement system. When a downstream sink fails or the source crashes, events between the last checkpoint and the crash point are lost — the checkpoint was advanced before delivery was confirmed.

The file source already supports acknowledgements using the same OrderedFinalizer + BatchNotifier pattern. This PR brings the same capability to kubernetes_logs, which shares the underlying file_source infrastructure.

Approach

The implementation mirrors the file source's acknowledgement pattern:

1. can_acknowledge() returns true — enables the topology's ack propagation
2. OrderedFinalizer<FinalizerEntry> — receives ack status from downstream sinks in order
3. BatchNotifier per batch — attached to events before emitting; downstream sinks update status on delivery
4. Checkpoint gated on ack — checkpoints.update() only called when BatchStatus::Delivered is received
5. Graceful shutdown — separate shutdown signal ties the finalizer stream to the checkpoint writer, ensuring all pending acks are processed before stopping

Key design decisions:

• FinalizerEntry is defined locally (not imported from file source) since sources-kubernetes_logs doesn't depend on sources-file
• Source::new_test() method added for mock-based testing with a pre-built Kubernetes client
• The acknowledgements config field uses the standard SourceAcknowledgementsConfig + bool_or_struct pattern

Vector configuration

sources:
  kubernetes_logs:
    type: kubernetes_logs
    # Acknowledgements are controlled at the sink level.
    # When any downstream sink has acknowledgements enabled,
    # the source automatically participates.

No source-level configuration is needed. Acknowledgements activate automatically when a downstream sink has acknowledgements.enabled = true, via propagate_acknowledgements().

How did you test this PR?

3 new tests + 51 existing tests pass:

Test	What it proves
file_start_position_server_restart_with_file_rotation_no_acknowledge	Without acks: checkpoint advances immediately, no re-read after restart
file_start_position_server_restart_with_file_rotation_acknowledged	With acks: checkpoint advances after ack, no duplicates after restart
checkpoint_does_not_advance_without_ack	Core safety: rejected events → checkpoint does NOT advance → data re-read after restart

Tests use a mock Kubernetes client (Source::new_test()) with a configurable logs directory, following the same pattern as the file source tests.

All 51 existing kubernetes_logs tests pass unchanged.

Change Type

• Bug fix
• New feature
• Dependencies
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

• Prior implementation: enhancement(kubernetes_logs source): Wire up acknowledgement support via existing file ack support vectordotdev/vector#22366