{"record":{"id":"548a4f77a0acb615","repo":"risingwavelabs/risingwave","slug":"confluent-schema-registry-error-error-code-mes","errorCode":null,"errorMessage":"confluent schema registry error {error_code}: {message}","messagePattern":"confluent schema registry error (.+?): (.+?)","errorType":"http","errorClass":"ErrorResp","httpStatus":null,"severity":"error","filePath":"src/connector/src/schema/schema_registry/util.rs","lineNumber":182,"sourceCode":"#[derive(Debug, Deserialize)]\npub struct GetByIdResp {\n    pub schema: String,\n}\n\n#[derive(Debug, Deserialize)]\npub struct GetBySubjectResp {\n    pub id: i32,\n    pub schema: String,\n    pub version: i32,\n    pub subject: String,\n    // default to empty/non-reference\n    #[serde(default)]\n    pub references: Vec<SchemaReference>,\n}\n\n/// <https://docs.confluent.io/platform/7.5/schema-registry/develop/api.html#errors>\n#[derive(Debug, Deserialize, thiserror::Error)]\n#[error(\"confluent schema registry error {error_code}: {message}\")]\npub struct ErrorResp {\n    error_code: i32,\n    message: String,\n}\n\n#[cfg(test)]\nmod test {\n    use super::super::handle_sr_list;\n\n    #[test]\n    fn test_handle_sr_list() {\n        let addr1 = \"http://localhost:8081\".to_owned();\n        assert_eq!(\n            handle_sr_list(&addr1).unwrap(),\n            vec![\"http://localhost:8081\".parse().unwrap()]\n        );\n\n        let addr2 = \"http://localhost:8081,http://localhost:8082\".to_owned();","sourceCodeStart":164,"sourceCodeEnd":200,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/schema/schema_registry/util.rs#L164-L200","documentation":"ErrorResp is the deserialized error body returned by the Confluent Schema Registry REST API; as a thiserror error it renders as 'confluent schema registry error {error_code}: {message}'. It surfaces a server-side registry error (e.g. 40401 SubjectNotFound, 409 Conflict, 42201 Invalid schema) with the registry's own error code and message.","triggerScenarios":"Any registry HTTP call that receives a non-success JSON error body which deserializes into ErrorResp — e.g. requesting a schema ID not present in the registry, registering an incompatible schema, or referencing an unknown subject/version.","commonSituations":"Schema ID from Kafka messages does not exist in the configured registry (wrong registry or registry recreated), schema incompatibility on registration, subject deleted (soft delete), or API auth returning registry-level errors.","solutions":["Look up the numeric error_code in the Confluent API docs to identify the exact condition (40401 subject not found, 40901 incompatible schema, etc.).","Verify the client points at the same registry that the producer registered schemas with.","Check the subject exists: GET /subjects and GET /subjects/{subject}/versions.","If the registry lost the schema (recreated), re-register the schema or repopulate IDs; fix incompatibilities by enabling/adjusting compatibility settings."],"exampleFix":"// before: wrong registry in sink/source\nschema.registry = 'http://new-registry:8081'  # schema IDs missing here\n// after: point at the registry holding the schemas\nschema.registry = 'http://original-registry:8081'","handlingStrategy":"try-catch","validationCode":"// pre-check subject exists before operations\nlet subjects: Vec<String> = reqwest::get(format!(\"{}/subjects\", registry_url))\n    .await?.json().await?;\nassert!(subjects.contains(&subject), \"subject {} not found\", subject);","typeGuard":null,"tryCatchPattern":"match req_inner(&ctx).await {\n    Ok(v) => v,\n    Err(RequestError::Unsuccessful(resp)) => {\n        tracing::error!(code = resp.error_code, msg = %resp.message, \"registry rejected request\");\n        match resp.error_code {\n            40401 => handle_subject_not_found(),\n            _ => return Err(resp.into()),\n        }\n    }\n    Err(e) => return Err(e.into()),\n}","preventionTips":["Map Confluent error codes (per API docs) to targeted handling.","Point consumers at the same registry the producers use.","Check subject existence and compatibility policy before registering schemas.","Avoid deleting/recreating registries that hold live schema IDs."],"tags":["schema-registry","api-error","kafka"],"backgroundTag":"api-error-response","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"}