{"record":{"id":"5ecbcdaa21f52ecb","repo":"vectordotdev/vector","slug":"partition-assignment-received-after-completion","errorCode":null,"errorMessage":"Partition assignment received after completion.","messagePattern":"Partition assignment received after completion\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":835,"sourceCode":"                        if !exit_eof {\n                            debug!(\"Partition consumer task finished, while not in draining mode.\");\n                        }\n                        state.keep_consuming(drain_deadline)\n                    },\n                };\n\n                // PartitionConsumerStatus differentiates between a task that exited after\n                // being signaled to end, and one that reached the end of its partition and\n                // was configured to exit. After the last such task ends, we signal the kafka\n                // driver task to shut down the main consumer too. Note this is only used in tests.\n                if exit_eof && status == PartitionConsumerStatus::PartitionEOF && partition_consumers.is_empty() {\n                    debug!(\"All partitions have exited or reached EOF.\");\n                    let _ = eof.take().map(|e| e.send(()));\n                }\n            },\n            Some(callback) = callbacks.recv() => match callback {\n                KafkaCallback::PartitionsAssigned(mut assigned_partitions, done) => match consumer_state {\n                    ConsumerState::Complete => unreachable!(\"Partition assignment received after completion.\"),\n                    ConsumerState::Draining(_) => error!(\"Partition assignment received while draining revoked partitions, maybe an invalid assignment.\"),\n                    ConsumerState::Consuming(ref consumer_state) => {\n                        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                    }","sourceCodeStart":817,"sourceCodeEnd":853,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L817-L853","documentation":"After the Kafka consumer reaches ConsumerState::Complete, the callback channel is expected closed, so no further KafkaCallback deliveries can arrive. A PartitionsAssigned callback hit in Complete means assignments were still delivered after completion - an unreachable state machine violation - so it panics instead of corrupting consumption state.","triggerScenarios":"A rebalance assignment racing the transition to Complete: the driver finalized (shutdown drain done, or EOF-mode completion) while the rdkafka rebalance callback thread still delivered an assignment; also possible when the callback channel wasn't closed on a given version's completion path.","commonSituations":"Shutdown coinciding with cluster rebalances (scaling events, broker leader changes, k8s pod evictions); sticky assignor churn; Vector/rdkafka version drift changing callback delivery timing.","solutions":["Upgrade Vector - completion/callback-close ordering received fixes","Reduce shutdown/rebalance overlap: generous graceful_shutdown_timeout and staggered instance restarts","Capture the callback sequence in debug logs (assignment arriving after 'All partitions have exited') and report it","Review exit_eof usage for that workload - continuous consumption avoids the EOF-completion path entirely"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Cannot be pre-validated; supervise the source task and restart on panic:\nif let Err(je) = tokio::spawn(kafka::run(cfg, shutdown)).await {\n    if je.is_panic() { /* log JoinError payload, alert, restart with backoff */ }\n}","preventionTips":["Stagger restarts and scale groups slowly so rebalances don't overlap shutdown","Prefer cooperative-sticky assignors to shrink rebalance windows","Keep Vector and the bundled rdkafka/librdkafka versions matched via official builds"],"tags":["rust","vector","kafka","rdkafka","rebalance","state-machine","shutdown","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"}