{"record":{"id":"06c031211b6890c6","repo":"risingwavelabs/risingwave","slug":"hummock-committed-epoch-sender-closed-unexpectedly","errorCode":null,"errorMessage":"Hummock committed epoch sender closed unexpectedly","messagePattern":"Hummock committed epoch sender closed unexpectedly","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":185,"sourceCode":"    ) -> anyhow::Result<(u64, Option<Vec<u8>>, Option<PbSinkSchemaChange>)> {\n        loop {\n            let wait_backoff = async {\n                if self.prepared_epochs.is_empty() {\n                    pending::<()>().await;\n                } else if let Some((backoff_fut, _)) = &mut self.backoff_state {\n                    backoff_fut.await;\n                }\n            };\n\n            select! {\n                _ = wait_backoff => {\n                    let item = self.prepared_epochs.front().cloned().expect(\"non-empty\");\n                    return Ok(item);\n                }\n\n                recv_epoch = self.job_committed_epoch_rx.recv() => {\n                    let Some(recv_epoch) = recv_epoch else {\n                        return Err(anyhow!(\n                            \"Hummock committed epoch sender closed unexpectedly\"\n                        ));\n                    };\n                    self.curr_hummock_committed_epoch = recv_epoch;\n                    while let Some((epoch, metadata, schema_change)) = self.pending_epochs.pop_front_if(|(epoch, _, _)| *epoch <= recv_epoch) {\n                        if let Some((last_epoch, _, _)) = self.prepared_epochs.back() {\n                            assert!(epoch > *last_epoch, \"prepared epochs must be in increasing order\");\n                        }\n                        self.prepared_epochs.push_back((epoch, metadata, schema_change));\n                    }\n                }\n            }\n        }\n    }\n\n    fn push_new_item(\n        &mut self,\n        epoch: u64,","sourceCodeStart":167,"sourceCodeEnd":203,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L167-L203","documentation":"Returned by `next_to_commit` when the `job_committed_epoch_rx` channel (carrying Hummock committed epochs from the observer manager) yields `None`, meaning all senders were dropped. Without a live committed-epoch feed the sink coordinator cannot make progress and errors out instead of hanging.","triggerScenarios":"The sender side of `job_committed_epoch_rx` (e.g. `observe_committed_epoch` feed in the meta node) is closed or dropped before/while `next_to_commit` polls it — typically when the meta service shuts down or the observer task terminates.","commonSituations":"Meta-node graceful shutdown while a sink coordinator is still running; a panic or early return in the task holding the sender; misconfigured startup where the committed-epoch observer never spawns.","solutions":["Check meta-node logs for the shutdown/crash of the component feeding job_committed_epoch (observer manager / hummock manager).","Ensure the coordinator worker task is cancelled before its epoch-feed sender is dropped, in the right shutdown order.","If the sender task crashed, find and fix its panic, then restart the meta node.","Verify the sink coordinator is only started after the committed-epoch channel is wired up."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// ensure the committed-epoch feed is alive before running the coordinator loop\nif job_committed_epoch_tx.is_closed() { return Err(anyhow!(\"committed epoch feed not started\")); }","typeGuard":null,"tryCatchPattern":"match worker.next_to_commit().await {\n    Err(e) if e.to_string().contains(\"sender closed\") => {\n        tracing::warn!(\"hummock committed epoch feed closed; stopping sink coordinator\");\n        // trigger orderly shutdown instead of retrying\n    }\n    other => other?,\n}","preventionTips":["Start the committed-epoch observer before spawning sink coordinator workers","Enforce shutdown ordering: cancel the coordinator before dropping the epoch-feed sender","Watch meta-node logs for panics in the sender task"],"tags":["meta","sink","channel","shutdown"],"backgroundTag":"broken-pipe","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}