{"record":{"id":"63a7a53897a842c1","repo":"vectordotdev/vector","slug":"partition-consumer-finished-after-completion","errorCode":null,"errorMessage":"Partition consumer finished after completion.","messagePattern":"Partition consumer finished after completion\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":802,"sourceCode":"    // should completely drain its acknowledgement stream after receiving an end signal\n    let mut partition_consumers: JoinSet<(TopicPartition, PartitionConsumerStatus)> =\n        Default::default();\n\n    // Handles that will let us end any consumer task that exceeds a drain deadline\n    let mut abort_handles: HashMap<TopicPartition, tokio::task::AbortHandle> = HashMap::new();\n\n    let exit_eof = eof.is_some();\n\n    while let ConsumerState::Consuming(_) | ConsumerState::Draining(_) = consumer_state {\n        tokio::select! {\n            Some(Ok((finished_partition, status))) = partition_consumers.join_next(), if !partition_consumers.is_empty() => {\n                debug!(\"Partition consumer finished for {}:{}\", &finished_partition.0, finished_partition.1);\n                // If this task ended on its own, the end_signal for it will still be in here.\n                end_signals.remove(&finished_partition);\n                abort_handles.remove(&finished_partition);\n\n                (drain_deadline, consumer_state) = match consumer_state {\n                    ConsumerState::Complete => unreachable!(\"Partition consumer finished after completion.\"),\n                    ConsumerState::Draining(mut state) => {\n                        state.partition_drained(finished_partition);\n\n                        if state.is_drain_complete() {\n                            debug!(\"All expected partitions have drained.\");\n                            state.finish_drain(drain_deadline)\n                        } else {\n                            state.keep_draining(drain_deadline)\n                        }\n                    },\n                    ConsumerState::Consuming(state) => {\n                        // If we are here, it is likely because the consumer\n                        // tasks are set up to exit upon reaching the end of the\n                        // partition.\n                        if !exit_eof {\n                            debug!(\"Partition consumer task finished, while not in draining mode.\");\n                        }\n                        state.keep_consuming(drain_deadline)","sourceCodeStart":784,"sourceCodeEnd":820,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L784-L820","documentation":"The Kafka source driver moves through ConsumerState::Consuming -> Draining -> Complete. Finished partition tasks arriving via join_next() are routed according to that state; a task finishing when the state machine is already Complete contradicts the design (no tasks should remain), hence unreachable!('Partition consumer finished after completion.')","triggerScenarios":"A partition consumer task outliving the driver's transition to Complete - a join_next() result delivered after the state was finalized, or an aborted task's result racing the completion path; a shutdown/rebalance race inside the select! loop.","commonSituations":"Heavy rebalance churn during shutdown (consumer group membership flapping), ack finalizers delaying task exit past the drain deadline, or rdkafka callback ordering changes across versions; typically intermittent and load-dependent.","solutions":["Upgrade Vector - the consumer state machine received fixes around shutdown races","Reduce rebalance churn: unique group id per instance, sane session.timeout.ms/heartbeat.interval.ms, stagger rolling restarts","Reproduce with debug logging for the kafka module and note the timing between 'Partition consumer finished' and completion","Report with logs if it recurs on the latest release"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Invariant violation inside Vector's driver task: contain it at the process level.\n// systemd 'Restart=on-failure' + RUST_BACKTRACE=1; in-process supervision:\nif let Err(je) = tokio::spawn(run_kafka_source(cfg, shutdown)).await {\n    if je.is_panic() { /* emit metric/alert, restart source with backoff */ }\n}","preventionTips":["Treat recurring unreachable! panics as bugs: collect logs plus configs and report them","Stagger rolling restarts of consumers sharing a group to limit rebalances","Monitor consumer-group lag and rebalance rate to catch the conditions that precede this race"],"tags":["rust","vector","kafka","rdkafka","shutdown","state-machine","panic","race-condition"],"backgroundTag":"kafka-shutdown-state-machine-violation","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}