{"record":{"id":"2e9380820e4e016a","repo":"apache/flink","slug":"could-not-find-schema-with-id-s-in-registry","errorCode":null,"errorMessage":"Could not find schema with id %s in registry","messagePattern":"Could not find schema with id (.+?) in registry","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java","lineNumber":76,"sourceCode":"     * @param schemaRegistryClient client to connect schema registry\n     */\n    public ConfluentSchemaRegistryCoder(SchemaRegistryClient schemaRegistryClient) {\n        this.schemaRegistryClient = schemaRegistryClient;\n    }\n\n    @Override\n    public Schema readSchema(InputStream in) throws IOException {\n        DataInputStream dataInputStream = new DataInputStream(in);\n\n        if (dataInputStream.readByte() != 0) {\n            throw new IOException(\"Unknown data format. Magic number does not match\");\n        } else {\n            int schemaId = dataInputStream.readInt();\n\n            try {\n                return schemaRegistryClient.getById(schemaId);\n            } catch (RestClientException e) {\n                throw new IOException(\n                        format(\"Could not find schema with id %s in registry\", schemaId), e);\n            }\n        }\n    }\n\n    @Override\n    public void writeSchema(Schema schema, OutputStream out) throws IOException {\n        try {\n            int registeredId = schemaRegistryClient.register(subject, schema);\n            out.write(CONFLUENT_MAGIC_BYTE);\n            byte[] schemaIdBytes = ByteBuffer.allocate(4).putInt(registeredId).array();\n            out.write(schemaIdBytes);\n        } catch (RestClientException e) {\n            throw new IOException(\"Could not register schema in registry\", e);\n        }\n    }\n}\n","sourceCodeStart":58,"sourceCodeEnd":94,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java#L58-L94","documentation":"IOException from ConfluentSchemaRegistryCoder.readSchema when schemaRegistryClient.getById(schemaId) throws RestClientException: the id encoded in the message cannot be resolved to a schema by the configured Schema Registry. Usually the registry is wrong/unreachable or the schema id was deleted/never existed there.","triggerScenarios":"Reading Confluent-wire-format messages whose schema id is unknown to the configured registry URL; pointing at a different registry environment (prod vs test); schema deleted by registry cleanup; transient registry HTTP errors surface as the same RestClientException.","commonSituations":"Environment mismatch between producing and consuming clusters; registry URL typo; auth (basic auth/TLS) not configured so registry returns 401/403; soft-deleted schemas after subject deletion.","solutions":["Verify the schema id exists: curl -u user:pass <registry-url>/schemas/ids/<id>.","Confirm 'avro-confluent-registry.schema-registry.url' (plus optional basic-auth credentials) points to the same registry the producer registered against.","If the schema was deleted, restore/re-register it under the original id or re-produce the data.","For transient registry outages, restart the job — Flink will retry from the last checkpoint."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// pre-flight: can the configured registry resolve the schema ids you expect?\ntry (CachedSchemaRegistryClient c = new CachedSchemaRegistryClient(url, 100)) {\n    Schema s = c.getById(expectedSchemaId); // throws if unknown\n} catch (RestClientException e) {\n    throw new IllegalStateException(\"Schema id \" + expectedSchemaId + \" not resolvable\", e);\n}","typeGuard":null,"tryCatchPattern":"try {\n    ... // deserialization loop\n} catch (IOException e) {\n    if (e.getMessage() != null && e.getMessage().startsWith(\"Could not find schema\")) {\n        // registry/env mismatch: check URL, auth, and id existence; do not skip the record\n        throw new IllegalStateException(\"Fatal: schema id unknown to registry\", e);\n    }\n    throw e;\n}","preventionTips":["Use the same registry (and credentials) for consumers as the producer registered against.","Disable schema deletion in the registry, or never delete schemas still referenced by retained messages.","Add a pre-deploy check that resolves the current max schema id from the registry."],"tags":["avro","confluent","schema-registry","network","kafka","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}