feat(nats source): Support End-to-End Acknowledgements (BatchNotifier) for JetStream

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

Currently, the nats source in Vector does not support End-to-End Acknowledgements (domain: acknowledgements). When pulling messages from a NATS JetStream stream, Vector immediately acknowledges the message back to the NATS broker as soon as it enters the local topology.

If a downstream sink (e.g., clickhouse, s3, elasticsearch) fails to write the events due to network errors, schema validation failures, or an in-flight process crash/OOM, Vector cannot request a redelivery from JetStream.

For setups relying on JetStream workqueue retention or strict durability constraints, this results in silent data loss: JetStream deletes the message upon early ACK, but the events never land in the destination sink.

Use Cases

I am running a high-throughput ingestion pipeline:
NATS JetStream -> Vector (Remap / Array Unnesting) -> ClickHouse

We rely on JetStream as a durable buffer. However, because Vector ACKs NATS messages at the topology boundary rather than after the ClickHouse HTTP batch write succeeds, any sink backpressure or container restart drops events permanently.

While draft PRs (like #26217) have attempted basic BatchNotifier wiring, an enterprise-ready implementation needs to account for several real-world pipeline edge cases detailed below.

Attempted Solutions & Alternatives Considered

Before opening this feature request, here is everything i have spent sleepless nights for the sake of JetStream durability constraint:

1. Benthos / Redpanda Connect (Tested & Verified)

• Configuration: Configured a nats_jetstream input pipeline writing directly to a clickhouse output.
• Result: Works natively as expected. Benthos holds the JetStream ACK until the ClickHouse HTTP driver returns a 200 OK, providing strict end-to-end durability without data loss during sink degradation.
• Why Vector is preferred: While Benthos handles the ACK loop, our infrastructure is standardized on Vector's topology management, VRL transformation ecosystem, and observability exporter suite. Migrating away from Vector solely for this source-level limitation adds unwanted architectural fragmentation.

2. Custom Node.js Ingestion Worker (Tested)

• Approach: Built a custom worker service using @nats-io/transport-node and the official ClickHouse JS client.
• Result: Provides explicit ACK control after batch flush, but introduces unnecessary maintenance overhead for standard stream processing logic (managing memory pressure, batching heuristics, and retry backoffs manually).

3. Native Vector Features (acknowledgements.enabled = true on ClickHouse Sink)

• Configuration:

  [sinks.clickhouse_out.acknowledgements]
  enabled = true
  
  
  

Proposed Solution & Requirements

We need the nats source to participate in Vector's vector_lib::event::BatchNotifier framework when acknowledgements.enabled = true.

To ensure this works reliably under production loads, the implementation should cover the following requirements:

1. Integration with BatchNotifier

Attach a BatchNotifier / EventFinalizer to the constructed Vector Event upon consuming an async_nats::jetstream::Message. A background task should await the BatchStatus from downstream sinks:

• BatchStatus::Successful: Send message.ack().await back to JetStream.
• BatchStatus::Failed: Send message.ack_with(AckKind::Nak).await (NACK) or allow the ack_wait timeout to handle redelivery based on source configuration.

2. Robust Handling for VRL Array Explosions (unnest!)

Pipelines frequently receive single NATS messages containing JSON arrays and explode them using VRL unnest!.

• Requirement: The source parent NATS message must only be ACKed when all child events derived from it successfully reach their respective sinks. Vector's reference-counting on BatchNotifier should be strictly verified through VRL topology steps.

3. Out-of-Order Batch ACK Management

Sinks (like ClickHouse) execute parallel async batch HTTP requests. Batch #2 might complete before Batch #1.

• Requirement: Ensure Tokio task management for in-flight selective ACKs handles non-sequential sink resolution without leaking memory or exceeding JetStream's ack_pending limits.

4. Graceful Shutdown / Drain Signalling

When Vector receives SIGTERM / SIGINT, topology sinks flush pending events before exiting.

• Requirement: The NATS ACK listener tasks must be bound to Vector’s shutdown drain signal to ensure final sink confirmations are sent back to the NATS server before Tokio runtime terminates.

5. Configurable Redelivery / NACK Backoff Strategy

Avoid hardcoded immediate Nak responses on transient sink failures.

• Requirement: If ClickHouse drops connection for 10 seconds, immediately sending Nak causes NATS to rapidly flood Vector with redeliveries, burning through the consumer's max_deliver threshold within seconds. A configurable backoff or defaulting to ack_wait expiration is necessary.

Configuration Example

[sources.nats_in]
type = "nats"
url = "nats://x.x.x.x:4222"
stream = "s2w-analytics"
consumer = "vector-clickhouse-sync"

[sources.nats_in.acknowledgements]
enabled = true

[sinks.clickhouse_out]
type = "clickhouse"
inputs = ["nats_in"]
endpoint = "http://x.x.x.x:8123"
database = "analytics"
table = "events"

[sinks.clickhouse_out.acknowledgements]
enabled = true

### References

#26217

### Version

0.57.0

Start with the nats source's JetStream message handling and Vector's BatchNotifier/EventFinalizer flow, then review the attempted approach in #26217. Trace how acknowledgements behave through VRL array expansion and downstream ClickHouse batches. Done means successful child-event completion ACKs the parent, failures permit safe redelivery, and shutdown drains pending confirmations without exceeding JetStream limits.