{"record":{"id":"184cf20f601b7179","repo":"apache/flink","slug":"schema-provided-for-s-format-does-not-match-the","errorCode":null,"errorMessage":"Schema provided for '%s' format does not match the table schema: %s","messagePattern":"Schema provided for '(.+?)' format does not match the table schema: (.+?)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java","lineNumber":256,"sourceCode":"                .getOptional(BEARER_AUTH_TOKEN)\n                .ifPresent(v -> properties.put(\"bearer.auth.token\", v));\n\n        if (properties.isEmpty()) {\n            return null;\n        }\n        return properties;\n    }\n\n    private static Schema getAvroSchema(String schemaString, RowType rowType) {\n        LogicalType convertedDataType =\n                AvroSchemaConverter.convertToDataType(schemaString).getLogicalType();\n\n        if (convertedDataType.isNullable()) {\n            convertedDataType = convertedDataType.copy(false);\n        }\n\n        if (!convertedDataType.equals(rowType)) {\n            throw new IllegalArgumentException(\n                    format(\n                            \"Schema provided for '%s' format does not match the table schema: %s\",\n                            IDENTIFIER, schemaString));\n        }\n\n        return new Parser().parse(schemaString);\n    }\n}\n","sourceCodeStart":238,"sourceCodeEnd":265,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java#L238-L265","documentation":"IllegalArgumentException from RegistryAvroFormatFactory.getAvroSchema: when 'avro-confluent-registry.schema' is provided, its converted logical type (after forcing non-nullable) must exactly equal the table's row type. Any difference in field names, order, types, nesting, or nullability triggers this mismatch.","triggerScenarios":"Setting 'format.schema' on an avro-confluent-registry table where the Avro schema's fields differ from the DDL columns; Avro unions producing nullable fields the DDL declares NOT NULL (or vice versa); column order differing between schema and DDL.","commonSituations":"Copying a producer's Avro schema into the DDL while the CREATE TABLE columns were reordered or retyped; schema drift after the registry schema evolved.","solutions":["Diff the two row types: print AvroSchemaConverter.convertToDataType(schemaString) and the table's derived row type — mismatched field, order, or nullability will show immediately.","Adjust the DDL so field names, order, types, and nullability exactly match the Avro schema (top-level nullability is ignored, nested nullability is not).","Alternatively drop 'format.schema' and let Flink derive the schema from the table definition."],"exampleFix":"-- before: schema says {'name':'id','type':['null','string']} but DDL had id BIGINT\n-- after\nCREATE TABLE t (id STRING, ...) WITH ('format'='avro-confluent-registry', 'avro-confluent-registry.schema'='{\"type\":\"record\",\"name\":\"r\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"string\"]}]}', ...)","handlingStrategy":"validation","validationCode":"// verify before submitting DDL\nDataType fromSchema =\n    org.apache.flink.formats.avro.typeutils.AvroSchemaConverter.convertToDataType(schemaString);\nLogicalType a = fromSchema.getLogicalType().copy(false); // force not-null like the factory does\nLogicalType b = tableRowType;\nif (!a.equals(b)) {\n    throw new IllegalArgumentException(\n        \"format.schema does not equal table row type:\\n schema=\" + a.asSummaryString()\n        + \"\\n table=\" + b.asSummaryString());\n}","typeGuard":null,"tryCatchPattern":"try {\n    tableEnv.executeSql(createSql);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"does not match the table schema\")) {\n        // compare AvroSchemaConverter.convertToDataType(schemaString) against DESCRIBE output\n    }\n    throw e;\n}","preventionTips":["When supplying 'format.schema', derive the DDL columns from the schema (names, order, nested nullability) mechanically rather than by hand.","Remember unions ['null', T] map to nullable fields — matching DDL columns must be nullable too.","Prefer omitting 'format.schema' and letting Flink derive it from the table when possible."],"tags":["avro","confluent","schema-mismatch","validation","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}