{"record":{"id":"c1bf05e5c54f4aec","repo":"risingwavelabs/risingwave","slug":"dynamic-channel-send-error-on-shutdown-signal","errorCode":null,"errorMessage":"(dynamic: channel send error on shutdown signal)","messagePattern":"\\(dynamic: channel send error on shutdown signal\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/redshift.rs","lineNumber":717,"sourceCode":"    async fn commit_schema_change(\n        &mut self,\n        _epoch: u64,\n        schema_change: PbSinkSchemaChange,\n    ) -> Result<()> {\n        use risingwave_pb::stream_plan::sink_schema_change::PbOp as SinkSchemaChangeOp;\n        let schema_change_op = schema_change\n            .op\n            .ok_or_else(|| SinkError::Coordinator(anyhow!(\"Invalid schema change operation\")))?;\n        let SinkSchemaChangeOp::AddColumns(add_columns) = schema_change_op else {\n            return Err(SinkError::Coordinator(anyhow!(\n                \"Only AddColumns schema change is supported for Redshift sink\"\n            )));\n        };\n        if let Some(shutdown_sender) = &self.shutdown_sender {\n            // Send shutdown signal to the periodic task before altering the table\n            shutdown_sender\n                .send(())\n                .map_err(|e| SinkError::Config(anyhow!(e)))?;\n        }\n        let sql = build_alter_add_column_sql(\n            self.config.schema.as_deref(),\n            &self.config.table,\n            &add_columns\n                .fields\n                .iter()\n                .map(|f| {\n                    let dt = DataType::from(f.data_type.as_ref().unwrap());\n                    Ok((f.name.clone(), convert_redshift_data_type(&dt)?))\n                })\n                .collect::<Result<Vec<_>>>()?,\n        );\n        let check_column_exists = |e: anyhow::Error| {\n            let err_str = e.to_report_string();\n            if regex::Regex::new(\".+ of relation .+ already exists\")\n                .unwrap()\n                .find(&err_str)","sourceCodeStart":699,"sourceCodeEnd":735,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/redshift.rs#L699-L735","documentation":"In `commit_schema_change`, before ALTERing the target table, the sink sends a shutdown signal over a bounded channel to stop the periodic task. `mpsc::Sender::send` fails only when the receiver is dropped, i.e., the periodic task has already terminated; the error is wrapped as a Config error.","triggerScenarios":"A schema change arrives when the periodic task has already exited (panicked or finished), so `shutdown_sender.send(())` returns SendError.","commonSituations":"Periodic task crashed earlier due to a DB error; sink schema change race with task termination; repeated schema changes after task shutdown.","solutions":["Investigate why the periodic task exited (check prior logs, e.g. panic reports)","Restart the sink so the periodic task and shutdown channel are recreated","Retry the schema change after the sink is healthy"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"if let Some(tx) = &self.shutdown_sender {\n    if tx.is_closed() { /* periodic task dead — restart before schema change */ }\n}","typeGuard":null,"tryCatchPattern":"if let Err(e) = shutdown_sender.send(()) {\n    // receiver dropped: periodic task already gone; log and continue/restart task\n    log::warn!(\"periodic task already stopped: {}\", e);\n}","preventionTips":["Check periodic task health before issuing schema changes","Recreate the sink if the background task has terminated"],"tags":["redshift","sink","channel","schema-change"],"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"}