What causes "(dynamic: channel send error on shutdown signal)" in risingwavelabs/risingwave?

Investigate why the periodic task exited (check prior logs, e.g. panic reports)

risingwavelabs/risingwave · error · SinkError::Config

(dynamic: channel send error on shutdown signal)

Error message

(dynamic: channel send error on shutdown signal)

What it means

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.

Source

Thrown at src/connector/src/sink/snowflake_redshift/redshift.rs:717

    async fn commit_schema_change(
        &mut self,
        _epoch: u64,
        schema_change: PbSinkSchemaChange,
    ) -> Result<()> {
        use risingwave_pb::stream_plan::sink_schema_change::PbOp as SinkSchemaChangeOp;
        let schema_change_op = schema_change
            .op
            .ok_or_else(|| SinkError::Coordinator(anyhow!("Invalid schema change operation")))?;
        let SinkSchemaChangeOp::AddColumns(add_columns) = schema_change_op else {
            return Err(SinkError::Coordinator(anyhow!(
                "Only AddColumns schema change is supported for Redshift sink"
            )));
        };
        if let Some(shutdown_sender) = &self.shutdown_sender {
            // Send shutdown signal to the periodic task before altering the table
            shutdown_sender
                .send(())
                .map_err(|e| SinkError::Config(anyhow!(e)))?;
        }
        let sql = build_alter_add_column_sql(
            self.config.schema.as_deref(),
            &self.config.table,
            &add_columns
                .fields
                .iter()
                .map(|f| {
                    let dt = DataType::from(f.data_type.as_ref().unwrap());
                    Ok((f.name.clone(), convert_redshift_data_type(&dt)?))
                })
                .collect::<Result<Vec<_>>>()?,
        );
        let check_column_exists = |e: anyhow::Error| {
            let err_str = e.to_report_string();
            if regex::Regex::new(".+ of relation .+ already exists")
                .unwrap()
                .find(&err_str)

View on GitHub (pinned to 6469eb736d)

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

Defensive patterns

Strategy: try-catch

Validate before calling

if let Some(tx) = &self.shutdown_sender {
    if tx.is_closed() { /* periodic task dead — restart before schema change */ }
}

Try / catch

if let Err(e) = shutdown_sender.send(()) {
    // receiver dropped: periodic task already gone; log and continue/restart task
    log::warn!("periodic task already stopped: {}", e);
}

Prevention

Check periodic task health before issuing schema changes
Recreate the sink if the background task has terminated

When it happens

Trigger: A schema change arrives when the periodic task has already exited (panicked or finished), so `shutdown_sender.send(())` returns SendError.

Common situations: Periodic task crashed earlier due to a DB error; sink schema change race with task termination; repeated schema changes after task shutdown.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/c1bf05e5c54f4aec. Report an issue: GitHub.