{"record":{"id":"da5ac04918da185c","repo":"risingwavelabs/risingwave","slug":"s3-configuration-is-required-for-redshift-s3-sink","errorCode":null,"errorMessage":"S3 configuration is required for redshift s3 sink","messagePattern":"S3 configuration is required for redshift s3 sink","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/redshift.rs","lineNumber":598,"sourceCode":"        mut shutdown_receiver: tokio::sync::mpsc::UnboundedReceiver<()>,\n    ) {\n        let mut copy_timer = interval(Duration::from_secs(write_intermediate_interval_seconds));\n        copy_timer.set_missed_tick_behavior(MissedTickBehavior::Skip);\n        let mut merge_timer = interval(Duration::from_secs(writer_target_interval_seconds));\n        merge_timer.set_missed_tick_behavior(MissedTickBehavior::Skip);\n\n        loop {\n            tokio::select! {\n                _ = shutdown_receiver.recv() => break,\n                _ = merge_timer.tick(), if merge_into_sql.is_some() => {\n                    if let Some(sql) = &merge_into_sql && let Err(e) = client.execute_sql_sync(sql.clone()).await {\n                        tracing::warn!(\"Failed to execute periodic query for table {}: {}\", config.table, e.as_report());\n                    }\n                },\n                _ = copy_timer.tick(), if need_copy_into => {\n                    if let Err(e) = async {\n                        let s3_inner = config.s3_inner.as_ref().ok_or_else(|| {\n                            SinkError::Config(anyhow!(\"S3 configuration is required for redshift s3 sink\"))\n                        })?;\n                        Self::flush_manifest_to_redshift(&client, &config,s3_inner, is_append_only).await?;\n                        Ok::<(),SinkError>(())\n                    }.await {\n                        tracing::error!(\"Failed to execute copy into task for sink id {}: {}\", sink_id, e.as_report());\n                    }\n                }\n            }\n        }\n        tracing::info!(\"Periodic query task stopped for sink id {}\", sink_id);\n    }\n}\n\nimpl Drop for RedshiftSinkCommitter {\n    fn drop(&mut self) {\n        // Send shutdown signal to the periodic task\n        if let Some(shutdown_sender) = &self.shutdown_sender\n            && let Err(e) = shutdown_sender.send(())","sourceCodeStart":580,"sourceCodeEnd":616,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/redshift.rs#L580-L616","documentation":"The periodic COPY task (`run_periodic_query_task`) requires S3 as the staging area between RisingWave and Redshift; this is thrown when `config.s3_inner` is None at the copy timer tick. `s3_inner` is populated only when the sink is configured with `with_s3`/S3 options, so this indicates the Redshift sink was configured for periodic COPY without valid S3 settings.","triggerScenarios":"The periodic copy task ticks (`need_copy_into == true`) but `s3_inner` is absent because the sink lacks S3 options or the S3 config failed to deserialize into `s3_inner`.","commonSituations":"Redshift sink DDL missing `s3.bucket`/`s3.path`/`s3.region`/credentials; typo in S3 option keys so they are not parsed; copy interval configured while S3 options omitted.","solutions":["Add the required S3 `with` options (s3.bucket, s3.path, s3.region, aws credentials) to the sink","Fix typo'd option keys so `s3_inner` gets populated","If COPY via S3 is not intended, remove the copy/s3-related options so the periodic copy task is not scheduled"],"exampleFix":"// before\nWITH ('connector'='redshift', 'copy.interval'='60s')\n// after\nWITH ('connector'='redshift', 'copy.interval'='60s',\n      's3.bucket'='my-bucket', 's3.path'='rw-sink', 's3.region'='us-east-1',\n      'aws.credentials.access_key_id'='...', 'aws.credentials.secret_access_key'='...')","handlingStrategy":"validation","validationCode":"if need_copy_into && config.s3_inner.is_none() {\n    return Err(\"S3 options are required when copy is enabled for Redshift sink\");\n}","typeGuard":"fn s3_ready(cfg: &RedshiftConfig) -> bool { cfg.s3_inner.is_some() }","tryCatchPattern":null,"preventionTips":["Pair every copy-interval/manifest option with full s3.* options","Double-check option key spelling ('s3.bucket', 's3.path', 's3.region')","Test sink creation against a MinIO endpoint before production"],"tags":["redshift","s3","config","sink"],"backgroundTag":"missing-required-config","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"}