{"record":{"id":"d890ac70228fcd28","repo":"risingwavelabs/risingwave","slug":"failed-to-receive-rebuild-sink-notify","errorCode":null,"errorMessage":"failed to receive rebuild sink notify","messagePattern":"failed to receive rebuild sink notify","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/sink.rs","lineNumber":897,"sourceCode":"                                    \"failed to rewind the log reader\"\n                                );\n                                Err(e)\n                            }\n                        }\n                    } else {\n                        Err(e)\n                    }\n                    .map_err(|e| StreamExecutorError::from((e, sink_param.sink_id)))?;\n                }\n            });\n            let message = loop {\n                select! {\n                    result = &mut future => {\n                        let Err(e): StreamExecutorResult<!> = result;\n                        return Err(e);\n                    }\n                    result = rebuild_sink_rx.recv() => {\n                        let message = result.ok_or_else(|| anyhow!(\"failed to receive rebuild sink notify\"))?;\n                        // Dropping the consumer costs a rewind, or a recovery when the log\n                        // reader cannot rewind. Not worth it for an update that changes nothing.\n                        if let RebuildSinkMessage::UpdateConfig(config) = &message\n                            && !sink_config_has_changes(&sink_param.properties, config)\n                        {\n                            info!(\n                                executor_id = %sink_writer_param.executor_id,\n                                sink_id = %sink_param.sink_id,\n                                \"skip alter sink config because properties are unchanged\"\n                            );\n                            continue;\n                        }\n                        break message;\n                    }\n                }\n            };\n            drop(future);\n            match message {","sourceCodeStart":879,"sourceCodeEnd":915,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/sink.rs#L879-L915","documentation":"While the sink executor consumes its write log (waiting on a rewind/rebuild future), it also listens on rebuild_sink_rx for messages from the writer (e.g. UpdateConfig from ALTER SINK CONFIG). If the channel closes with no message — the writer task is gone — recv returns None and the executor raises this error instead of silently proceeding.","triggerScenarios":"rebuild_sink_rx.recv() returns None while execute_consume_log is select!-waiting: the paired writer (execute_write_log side) dropped rebuild_sink_tx, typically because the write-log executor terminated.","commonSituations":"Writer task exited due to a prior error, actor cancellation racing with log consumption, or misrouted channel wiring after executor restart.","solutions":["Look for an earlier error/exit of the write-log executor in the same actor's logs","Restart the actor so writer and reader halves are re-created together","Verify channel pairing in the sink executor split (writer holds tx, consumer holds rx) hasn't been broken by recent changes","Escalate to RisingWave maintainers if reproducible — it signals an unexpected writer shutdown"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"let message = match rebuild_sink_rx.recv().await {\n    Some(m) => m,\n    None => return Err(anyhow!(\"failed to receive rebuild sink notify\")),\n};","preventionTips":["Keep writer and reader halves created and dropped together in the executor split","Log writer-side exits so a closed rx is explainable","Add tests where the writer exits while the consumer is select!-waiting"],"tags":["stream-executor","sink","channel"],"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"}