{"record":{"id":"8858a533c8a2cfe3","repo":"vectordotdev/vector","slug":"drain-deadline-received-after-completion","errorCode":null,"errorMessage":"Drain deadline received after completion.","messagePattern":"Drain deadline received after completion\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":923,"sourceCode":"                                    state.revoke_partition(tp, end);\n                                } _ => {\n                                    debug!(\"Consumer task for partition {}:{} already finished.\", &tp.0, tp.1);\n                                }}\n                            });\n                        }\n                        // If shutdown was initiated by partition EOF mode, the drain phase\n                        // will already be complete and would time out if not accounted for here\n                        if state.is_drain_complete() {\n                            state.finish_drain(deadline)\n                        } else {\n                            state.keep_draining(deadline)\n                        }\n                    }\n                },\n            },\n\n            Some(_) = &mut drain_deadline => (drain_deadline, consumer_state) = match consumer_state {\n                ConsumerState::Complete => unreachable!(\"Drain deadline received after completion.\"),\n                ConsumerState::Consuming(state) => {\n                    warn!(\"A drain deadline fired outside of draining mode.\");\n                    state.keep_consuming(None.into())\n                },\n                ConsumerState::Draining(mut draining) => {\n                    debug!(\"Acknowledgement drain deadline reached. Dropping any pending ack streams for revoked partitions.\");\n                    for tp in draining.consumer_state.expect_drain.drain() {\n                        if let Some(handle) = abort_handles.remove(&tp) {\n                            handle.abort();\n                        }\n                    }\n                    draining.finish_drain(drain_deadline)\n                }\n            },\n        }\n    }\n}\n","sourceCodeStart":905,"sourceCodeEnd":941,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L905-L941","documentation":"Drain deadlines (the max_drain_ms acknowledgement drain timeout) are armed when partitions are revoked/shutting down and fire in Draining state. If the deadline future wakes while the state is already Complete - deadlines should have been consumed or dropped at completion - the invariant is broken and it panics.","triggerScenarios":"The drain-deadline sleep resolving after the transition to Complete (timer not cancelled on the completion path), or a duplicated deadline future; both are shutdown-race shapes inside the driver's select! loop.","commonSituations":"Very short max_drain_ms combined with fast drains, tokio timer pressure under load, and versions where the deadline wasn't taken/dropped on the completion branch.","solutions":["Upgrade Vector - timer/branch handling around drain completion was fixed over releases","Give max_drain_ms comfortable margin above worst-case acknowledgement latency so the deadline rarely fires at the boundary","Capture debug logs ('Acknowledgement drain deadline reached' near completion) and report","If embedding, prefer graceful shutdown that awaits the source future instead of cancelling it mid-drain"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Timer/state race inside Vector: contain and restart the source task:\nif let Err(je) = tokio::spawn(kafka::run(cfg, shutdown)).await {\n    if je.is_panic() { /* log, restart with backoff; attach debug logs to a report */ }\n}","preventionTips":["Set max_drain_ms well above worst-case ack latency so deadlines don't fire at drain boundaries","Avoid cancelling the source future mid-drain; let graceful shutdown await it","Upgrade promptly - drain timer handling improves across releases"],"tags":["rust","vector","kafka","shutdown","timer","drain","panic","race-condition"],"backgroundTag":"kafka-drain-deadline-race","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}