vectordotdev/vector · error

Finalizer must have been set up for acknowledgements

Error message

Finalizer must have been set up for acknowledgements

What it means

In gcp_pubsub's streaming pull, each response batch sets up acknowledgement tracking with (batch, notifier) = BatchNotifier::maybe_new_with_receiver(self.acknowledgements). When the notifier exists, delivered ids are routed through finalizer.as_ref().expect("Finalizer must have been set up for acknowledgements"). The invariant: whenever acknowledgements are enabled (so notifier is Some), handle_response is called with a Some finalizer. The panic means notifier and finalizer were built from different assumptions - an internal wiring bug, not a GCP-side condition.

Source

Thrown at src/sources/gcp_pubsub.rs:642

            busy_flag.store(true, Ordering::Relaxed);
        }
        self.bytes_received.emit(ByteSize(response.size_of()));

        let (batch, notifier) = BatchNotifier::maybe_new_with_receiver(self.acknowledgements);
        let (events, ids) = self.parse_messages(response.received_messages, batch).await;

        let count = events.len();
        match self.out.send_batch(events).await {
            Err(_) => emit!(StreamClosedError { count }),
            Ok(()) => match notifier {
                None => ack_ids
                    .send(ids)
                    .await
                    .unwrap_or_else(|_| unreachable!("request stream never closes")),
                Some(notifier) => {
                    finalizer
                        .as_ref()
                        .expect("Finalizer must have been set up for acknowledgements")
                        .add(ids, notifier);
                    *pending_acks += 1;
                }
            },
        }
    }

    async fn parse_messages(
        &self,
        response: Vec<proto::ReceivedMessage>,
        batch: Option<BatchNotifier>,
    ) -> (Vec<Event>, Vec<String>) {
        let mut ack_ids = Vec::with_capacity(response.len());
        let events = response
            .into_iter()
            .flat_map(|received| {
                ack_ids.push(received.ack_id);
                received

View on GitHub (pinned to 3708c39b12)

Solutions

1. On stock Vector: report with config and version - this is an unreachable-invariant bug worth fixing structurally
2. Workaround: run the source with acknowledgements disabled if your pipeline semantics allow it
3. Patch: pass a plain Finalizer (not Option) when acks are on, or derive finalizer and notifier from a single Option so they cannot diverge

Example fix

// before
Some(notifier) => {
    finalizer
        .as_ref()
        .expect("Finalizer must have been set up for acknowledgements")
        .add(ids, notifier);
}

// after - derive both from one value so they cannot diverge
match (finalizer.as_ref(), notifier) {
    (Some(finalizer), Some(notifier)) => {
        finalizer.add(ids, notifier);
        *pending_acks += 1;
    }
    (None, Some(_)) => {
        error!(message = "acknowledgements enabled without finalizer; acking directly");
        ack_ids.send(ids).await.ok();
    }
    (_, None) => { ack_ids.send(ids).await.ok(); }
}

// make the two pieces impossible to diverge before starting the pull loop:
let (finalizer, acks_enabled) = if config acknowledgements {
    (Some(finalizer), true)
} else {
    (None, false)
};
debug_assert_eq!(finalizer.is_some(), acks_enabled, "finalizer must exist iff acknowledgements are enabled");

fn ack_pair(finalizer: &Option<Finalizer>, notifier: &Option<BatchNotifier>) -> bool {
    finalizer.is_some() || notifier.is_none()
}

match (finalizer.as_ref(), notifier) {
    (Some(f), Some(n)) => { f.add(ids, n); *pending_acks += 1; }
    (None, Some(_)) => { error!("acks enabled without finalizer"); ack_ids.send(ids).await.ok(); }
    (_, None) => { ack_ids.send(ids).await.ok(); }
}

When it happens

Trigger: Invoking the streaming-pull loop with acknowledgements enabled but without constructing the Finalizer - custom forks or refactors that create the BatchNotifier path but skip the finalizer setup. Stock Vector derives both from the same source config, so the branch should be unreachable.

Common situations: Custom builds between gcp_pubsub acknowledgement refactors; toggling acknowledgement settings in forks without rebuilding the finalizer path.

Related errors

• Indexer acknowledgements channel must allow at least one pen
• MessageStream never calls Ready(None)
• MessageStream never returns Ready(None)
• Messages are consumed in dedicated tasks for each partition.
• No timeout is configured for this source.

AI-assisted analysis of vectordotdev/vector@3708c39b12 (2026-08-20). Data as JSON: /api/errors/62e7901ae86b35bb. Report an issue: GitHub.