{"record":{"id":"46b8aec4dfa190d3","repo":"risingwavelabs/risingwave","slug":"unable-to-send-response","errorCode":null,"errorMessage":"unable to send response: {:?}","messagePattern":"unable to send response: (.+?)","errorType":"exception","errorClass":"SinkError::Remote","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/remote.rs","lineNumber":346,"sourceCode":"    async fn consume_log_and_sink(self, mut log_reader: impl SinkLogReader) -> Result<!> {\n        log_reader.start_from(None).await?;\n        let mut request_tx = self.request_sender;\n        let mut response_err_stream_rx = self.response_stream;\n        let sink_writer_metrics = self.sink_writer_metrics;\n\n        let (response_tx, mut response_rx) = unbounded_channel();\n\n        let poll_response_stream = async move {\n            loop {\n                let result = response_err_stream_rx\n                    .stream\n                    .try_next()\n                    .instrument_await(\"log_sinker_wait_next_response\")\n                    .await;\n                match result {\n                    Ok(Some(response)) => {\n                        response_tx.send(response).map_err(|err| {\n                            SinkError::Remote(anyhow!(\"unable to send response: {:?}\", err.0))\n                        })?;\n                    }\n                    Ok(None) => return Err(SinkError::Remote(anyhow!(\"end of response stream\"))),\n                    Err(e) => return Err(SinkError::Remote(anyhow!(e))),\n                }\n            }\n        };\n\n        let poll_consume_log_and_sink = async move {\n            fn truncate_matched_offset(\n                queue: &mut VecDeque<(TruncateOffset, Option<Instant>)>,\n                persisted_offset: TruncateOffset,\n                log_reader: &mut impl SinkLogReader,\n                sink_writer_metrics: &SinkWriterMetrics,\n            ) -> Result<()> {\n                while let Some((sent_offset, _)) = queue.front()\n                    && sent_offset < &persisted_offset\n                {","sourceCodeStart":328,"sourceCodeEnd":364,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/remote.rs#L328-L364","documentation":"In the log-sink writer loop, the remote sink tries to forward each response received from the remote writer stream to response_tx (the channel back to the stream executor). If the receiving end of that channel has already been dropped (receiver gone), send fails and this error is raised.","triggerScenarios":"consume_log_and_sink receives Ok(Some(response)) from the log reader, but response_tx.send fails because the downstream receiver task was dropped/aborted (e.g. executor shutdown, stream aborted before draining responses).","commonSituations":"Actor/executor is being shut down or migrated during recovery while the sink writer task is still pushing responses; a panic or early return in the consumer side; barrier aborts that drop the channel early.","solutions":["Typically benign during shutdown/recovery; check whether the sink actor was restarted and whether the error is recurring.","If recurring, investigate why the response receiver exits before the writer finishes (look at executor logs just before this error).","Retry the sink after recovery completes; RisingWave will rebuild the channel."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"if let Err(e) = result {\n    if e.to_string().contains(\"unable to send response\") {\n        // receiver dropped; treat as shutdown path, don't crash the actor\n        return Ok(()); // or propagate depending on recovery semantics\n    }\n    return Err(e);\n}","preventionTips":["Keep the response receiver alive until the writer task completes","Handle actor shutdown by cancelling the writer before dropping the channel","Treat this as expected during recovery/restart and verify via metrics that the sink resumes"],"tags":["channel","sink","async","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"}