apache/flink · error · IOException

Unknown data format. Magic number does not match

Error message

Unknown data format. Magic number does not match

What it means

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.

Source

Thrown at flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java:69

        this.subject = subject;
    }

    /**
     * Creates {@link SchemaCoder} that uses provided {@link SchemaRegistryClient} to connect to
     * schema registry.
     *
     * @param schemaRegistryClient client to connect schema registry
     */
    public ConfluentSchemaRegistryCoder(SchemaRegistryClient schemaRegistryClient) {
        this.schemaRegistryClient = schemaRegistryClient;
    }

    @Override
    public Schema readSchema(InputStream in) throws IOException {
        DataInputStream dataInputStream = new DataInputStream(in);

        if (dataInputStream.readByte() != 0) {
            throw new IOException("Unknown data format. Magic number does not match");
        } else {
            int schemaId = dataInputStream.readInt();

            try {
                return schemaRegistryClient.getById(schemaId);
            } catch (RestClientException e) {
                throw new IOException(
                        format("Could not find schema with id %s in registry", schemaId), e);
            }
        }
    }

    @Override
    public void writeSchema(Schema schema, OutputStream out) throws IOException {
        try {
            int registeredId = schemaRegistryClient.register(subject, schema);
            out.write(CONFLUENT_MAGIC_BYTE);
            byte[] schemaIdBytes = ByteBuffer.allocate(4).putInt(registeredId).array();

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Verify the data producer: it must use Confluent Schema Registry Avro serializer (magic byte 0 + schema id).
2. If data is plain Avro, switch the table format back to 'avro' and supply 'format.avro-schema'.
3. Inspect the first bytes of a sample message (hex) to identify the actual wire format.

Example fix

-- before
WITH ('format'='avro-confluent-registry', 'avro-confluent-registry.schema-registry.url'='...')
-- on a topic containing plain Avro bytes

-- after
WITH ('format'='avro', 'avro-schema'='...')

// sniff one message before starting the job
byte[] sample = readOneMessage(topic);
if ((sample[0] & 0xFF) != 0) {
    throw new IllegalArgumentException(
        "Topic does not contain Confluent wire-format Avro (magic byte != 0); use format 'avro'");
}

boolean isConfluentWireFormat(byte[] msg) { return msg != null && msg.length > 5 && (msg[0] & 0xFF) == 0; }

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Could not find schema with id %s in registry
• Failed to serialize schema registry.
• Could not register schema in registry
• Option %s.%s is required for serialization
• Failed to deserialize Avro record.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/a52a7b39a03a1bc6. Report an issue: GitHub.