{"record":{"id":"98d86781772f35f0","repo":"risingwavelabs/risingwave","slug":"sink-error-0-98d867","errorCode":null,"errorMessage":"Sink error: {0}","messagePattern":"Sink error: (.+?)","errorType":"error_code","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/error.rs","lineNumber":117,"sourceCode":"\n    #[error(\"System parameters error: {0}\")]\n    SystemParams(String),\n\n    #[error(\"Session parameters error: {0}\")]\n    SessionConfig(\n        #[from]\n        #[backtrace]\n        SessionConfigError,\n    ),\n\n    #[error(transparent)]\n    Connector(\n        #[from]\n        #[backtrace]\n        ConnectorError,\n    ),\n\n    #[error(\"Sink error: {0}\")]\n    Sink(\n        #[from]\n        #[backtrace]\n        SinkError,\n    ),\n\n    #[error(transparent)]\n    Internal(\n        #[from]\n        #[backtrace]\n        anyhow::Error,\n    ),\n\n    // Indicates that recovery was triggered manually.\n    #[error(\"adhoc recovery triggered\")]\n    AdhocRecovery,\n\n    #[error(\"Integrity check failed\")]","sourceCodeStart":99,"sourceCodeEnd":135,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/error.rs#L99-L135","documentation":"This error wraps a SinkError (from the sink/connector sink subsystem) into the meta node's unified MetaError. It is raised when meta-node code that manages sinks (creation, drops, streaming job scheduling, sink coordinator) propagates a failure that originated in a sink implementation (e.g. a connector sink writing to an external system). The `#[from]` attribute means any SinkError is automatically converted into MetaError::Sink via `?`.","triggerScenarios":"Any code path in the meta service that calls sink-related APIs and propagates a SinkError with `?` or From conversion: creating/starting a sink via the sink manager, sink coordinator writing to the external target, or sink schema validation failing during sink creation.","commonSituations":"External sink target (Kafka, Elasticsearch, Redis, etc.) is down or misconfigured; sink credentials are wrong; target schema changed so the sink can no longer write; network partition between meta/compute nodes and the sink system.","solutions":["Inspect the wrapped SinkError (source of this error) to find the underlying connector failure","Verify the sink connection options (bootstrap servers, URL, credentials) in the CREATE SINK statement","Check network connectivity from meta/compute nodes to the sink target","Recreate or restart the sink after fixing the external system","Check meta node logs for the full backtrace attached to this variant"],"exampleFix":"// before: blindly propagating\nlet result = sink_manager.write(msg)?;\n// after: match on the specific sink failure\nmatch sink_manager.write(msg) {\n    Err(MetaError::Sink(e)) => { tracing::error!(\"sink write failed: {e}\"); retry_or_pause(e) }\n    r => r,\n}","handlingStrategy":"try-catch","validationCode":"// Rust: check sink options before CREATE SINK\nasync fn ensure_sink_target_reachable(opts: &SinkOptions) -> Result<(), MetaError> {\n    connector::test_sink_connectivity(opts).await\n        .map_err(|e| MetaError::Sink(SinkError::from(e)))\n}","typeGuard":"fn is_sink_error(e: &MetaError) -> Option<&SinkError> {\n    match e { MetaError::Sink(s) => Some(s), _ => None }\n}","tryCatchPattern":"match meta_result {\n    Err(MetaError::Sink(sink_err)) => {\n        tracing::error!(\"sink failed: {sink_err:#}\");\n        // inspect sink_err.source() for the connector cause, then retry/backoff\n    }\n    other => other?,\n}","preventionTips":["Run connector connectivity checks (WITH options) before creating sinks","Keep sink credentials in secrets and rotate them proactively","Monitor sink target health (broker lag, endpoint availability)","Pin connector config to schemas supported by your RisingWave version"],"tags":["sink","connector","meta-node"],"backgroundTag":"upstream-api-error","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}