{"record":{"id":"3ed3b2f6c8d3de75","repo":"vectordotdev/vector","slug":"shutdown-received-after-completion","errorCode":null,"errorMessage":"Shutdown received after completion.","messagePattern":"Shutdown received after completion\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":880,"sourceCode":"                        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                        }\n\n                        state.keep_draining(deadline)\n                    }\n                },\n                KafkaCallback::ShuttingDown(drain) => (drain_deadline, consumer_state) = match consumer_state {\n                    ConsumerState::Complete => unreachable!(\"Shutdown received after completion.\"),\n                    // Shutting down is just like a full assignment revoke, but we also close the\n                    // callback channels, since we don't expect additional assignments or rebalances\n                    ConsumerState::Draining(state) => {\n                        // NB: This would only happen if the task driving the kafka client is\n                        // not handling shutdown signals; otherwise this is unreachable code\n                        error!(\"Kafka client handled a shutdown signal while a rebalance was in progress.\");\n                        callbacks.close();\n                        state.keep_draining(drain_deadline)\n                    },\n                    ConsumerState::Consuming(state) => {\n                        callbacks.close();\n                        let (deadline, mut state) = state.begin_drain(max_drain_ms, drain, true);\n                        if let Ok(tpl) = consumer.assignment() {\n                            // TODO  workaround for https://github.com/fede1024/rust-rdkafka/issues/681\n                            if tpl.capacity() == 0 {\n                                return;\n                            }\n                            tpl.elements()","sourceCodeStart":862,"sourceCodeEnd":898,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L862-L898","documentation":"KafkaCallback::ShuttingDown is rdkafka's pre-shutdown notification, handled in Consuming (start drain) and Draining (error + close callbacks) states. In Complete no callbacks are expected at all; receiving one means the kafka client thread signaled shutdown after the driver already finalized - unreachable, so it panics.","triggerScenarios":"The rdkafka client-level shutdown callback firing after the driver's own completion transition (e.g. EOF mode completed the consumer and the late client shutdown signal arrives); duplicate delivery of the shutting-down callback after channel close in a given rdkafka version.","commonSituations":"exit_eof workflows finishing exactly as the client shuts down; rapid restart cycles; rdkafka/librdkafka version mismatches in the deployment image.","solutions":["Upgrade Vector (shutdown callback ordering fixes)","If using exit_eof, verify the workload actually needs it - continuous consumption avoids this completion path","Reproduce with debug logs and report the callback ordering","Check linked librdkafka consistency in the image: ldd $(which vector) | grep rdkafka"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Process-level containment; the panic kills the source task by design:\n//   systemd: Restart=on-failure, Environment=RUST_BACKTRACE=1\nif let Err(je) = tokio::spawn(kafka::run(cfg, shutdown)).await {\n    if je.is_panic() { /* log, restart with backoff; report if recurring */ }\n}","preventionTips":["Avoid exit_eof where continuous consumption works - it removes this completion path","Keep graceful_shutdown_timeout long enough for clean drains","Verify librdkafka version consistency in custom images (ldd)"],"tags":["rust","vector","kafka","rdkafka","shutdown","callback","panic","race-condition"],"backgroundTag":"kafka-shutdown-race","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}