{"record":{"id":"a52a7b39a03a1bc6","repo":"apache/flink","slug":"unknown-data-format-magic-number-does-not-match","errorCode":null,"errorMessage":"Unknown data format. Magic number does not match","messagePattern":"Unknown data format\\. Magic number does not match","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":69,"sourceCode":"        this.subject = subject;\n    }\n\n    /**\n     * Creates {@link SchemaCoder} that uses provided {@link SchemaRegistryClient} to connect to\n     * schema registry.\n     *\n     * @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();","sourceCodeStart":51,"sourceCodeEnd":87,"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#L51-L87","documentation":"IOException from ConfluentSchemaRegistryCoder.readSchema when the first byte of the message is not the Confluent wire-format magic byte 0x0. Confluent-serialized Avro messages start with magic byte 0 plus a 4-byte schema id; any other leading byte means the payload was not produced with the Confluent wire format.","triggerScenarios":"Pointing an 'avro-confluent-registry' table at a topic containing plain (schema-less) Avro binary data, raw JSON, or Confluent JSON-encoded messages; reading a topic with mixed producers.","commonSituations":"Switching format from 'avro' to 'avro-confluent-registry' without re-producing the data; a legacy producer writing plain Avro into the same topic; accidentally reading a compacted-string topic.","solutions":["Verify the data producer: it must use Confluent Schema Registry Avro serializer (magic byte 0 + schema id).","If data is plain Avro, switch the table format back to 'avro' and supply 'format.avro-schema'.","Inspect the first bytes of a sample message (hex) to identify the actual wire format."],"exampleFix":"-- before\nWITH ('format'='avro-confluent-registry', 'avro-confluent-registry.schema-registry.url'='...')\n-- on a topic containing plain Avro bytes\n\n-- after\nWITH ('format'='avro', 'avro-schema'='...')","handlingStrategy":"validation","validationCode":"// sniff one message before starting the job\nbyte[] sample = readOneMessage(topic);\nif ((sample[0] & 0xFF) != 0) {\n    throw new IllegalArgumentException(\n        \"Topic does not contain Confluent wire-format Avro (magic byte != 0); use format 'avro'\");\n}","typeGuard":"boolean isConfluentWireFormat(byte[] msg) { return msg != null && msg.length > 5 && (msg[0] & 0xFF) == 0; }","tryCatchPattern":null,"preventionTips":["Match the table format to the producer serializer: Confluent registry serializer -> 'avro-confluent-registry', plain Avro -> 'avro'.","Hex-dump the first bytes of sample messages when onboarding a topic.","Keep one serialization format per topic."],"tags":["avro","confluent","schema-registry","wire-format","kafka","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}