{"record":{"id":"c19c7491073b6ffc","repo":"vectordotdev/vector","slug":"partitions-revoked-after-completion","errorCode":null,"errorMessage":"Partitions revoked after completion.","messagePattern":"Partitions revoked after completion\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":856,"sourceCode":"                        let acks = consumer.context().acknowledgements;\n                        for tp in assigned_partitions.drain(0..) {\n                            let topic = tp.0.as_str();\n                            let partition = tp.1;\n                            match consumer.split_partition_queue(topic, partition) { Some(pq) => {\n                                debug!(\"Consuming partition {}:{}.\", &tp.0, tp.1);\n                                let (end_tx, handle) = consumer_state.consume_partition(&mut partition_consumers, tp.clone(), Arc::clone(&consumer), pq, acks, exit_eof);\n                                abort_handles.insert(tp.clone(), handle);\n                                end_signals.insert(tp, end_tx);\n                            } _ => {\n                                warn!(\"Failed to get queue for assigned partition {}:{}.\", &tp.0, tp.1);\n                            }}\n                        }\n                        // ensure this is retained until all individual queues are set up\n                        drop(done);\n                    }\n                },\n                KafkaCallback::PartitionsRevoked(mut revoked_partitions, drain) => (drain_deadline, consumer_state) = match consumer_state {\n                    ConsumerState::Complete => unreachable!(\"Partitions revoked after completion.\"),\n                    ConsumerState::Draining(d) => {\n                        // NB: This would only happen if the task driving the kafka client (i.e. rebalance handlers)\n                        // is not handling shutdown signals, and a revoke happens during a shutdown drain; otherwise\n                        // this is unreachable code.\n                        warn!(\"Kafka client is already draining revoked partitions.\");\n                        d.keep_draining(drain_deadline)\n                    },\n                    ConsumerState::Consuming(state) => {\n                        let (deadline, mut state) = state.begin_drain(max_drain_ms, drain, false);\n\n                        for tp in revoked_partitions.drain(0..) {\n                            match end_signals.remove(&tp) { Some(end) => {\n                                debug!(\"Revoking partition {}:{}\", &tp.0, tp.1);\n                                state.revoke_partition(tp, end);\n                            } _ => {\n                                debug!(\"Consumer task for partition {}:{} already finished.\", &tp.0, tp.1);\n                            }}\n                        }","sourceCodeStart":838,"sourceCodeEnd":874,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L838-L874","documentation":"Symmetric to assignments: once the driver is Complete no KafkaCallback::PartitionsRevoked can legitimately arrive (callback channels are closed at completion). A revoke in Complete violates the shutdown state machine and panics - the alternative would be silently losing acknowledgement bookkeeping for already-finalized partitions.","triggerScenarios":"A revoke callback delivered after the state machine completed - shutdown drain finishing while a group rebalance or broker-initiated revoke is still in flight on the rdkafka callback thread; aborted-task cleanup racing the completion transition.","commonSituations":"Rolling restarts or k8s evictions during active rebalances; consumers sharing the group with different session timeouts; rdkafka version drift changing callback ordering.","solutions":["Upgrade Vector to pick up state machine race fixes","Minimize rebalances during shutdown: quiesce producers before restarting Vector, stagger restarts, prefer cooperative-sticky assignors","Collect debug logs showing the revoke-after-complete sequence and open an issue","As a workaround in flaky environments, run a single consumer per group so external rebalances cannot interleave"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Same containment shape: the driver task panic is fatal for the source,\n// so supervise and recover at the process/task boundary:\nif let Err(je) = tokio::spawn(kafka::run(cfg, shutdown)).await {\n    if je.is_panic() { /* log, alert, restart with exponential backoff */ }\n}","preventionTips":["Quiesce producers before restarting consumers to avoid revoke storms","Use a dedicated group id per deployment and aligned session timeouts across members","Capture debug logs of rebalance events to include in bug reports"],"tags":["rust","vector","kafka","rdkafka","rebalance","shutdown","state-machine","panic"],"backgroundTag":"kafka-rebalance-after-shutdown","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}