{"record":{"id":"5e5223cff5182e89","repo":"apache/flink","slug":"schema-provided-for-s-format-must-be-a-nullable","errorCode":null,"errorMessage":"Schema provided for '%s' format must be a nullable record type with fields 'before', 'after', 'op' and schema of fields 'before' and 'after' must match the table schema: %s","messagePattern":"Schema provided for '(.+?)' format must be a nullable record type with fields 'before', 'after', 'op' and schema of fields 'before' and 'after' must 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/debezium/DebeziumAvroFormatFactory.java","lineNumber":190,"sourceCode":"        options.add(SCHEMA);\n        options.add(SSL_KEYSTORE_LOCATION);\n        options.add(SSL_KEYSTORE_PASSWORD);\n        options.add(SSL_TRUSTSTORE_LOCATION);\n        options.add(SSL_TRUSTSTORE_PASSWORD);\n        options.add(BASIC_AUTH_CREDENTIALS_SOURCE);\n        options.add(BASIC_AUTH_USER_INFO);\n        options.add(BEARER_AUTH_CREDENTIALS_SOURCE);\n        options.add(BEARER_AUTH_TOKEN);\n        return options;\n    }\n\n    static void validateSchemaString(@Nullable String schemaString, RowType rowType) {\n        if (schemaString != null) {\n            LogicalType convertedDataType =\n                    AvroSchemaConverter.convertToDataType(schemaString).getLogicalType();\n\n            if (!convertedDataType.equals(rowType)) {\n                throw new IllegalArgumentException(\n                        format(\n                                \"Schema provided for '%s' format must be a nullable record type with fields 'before', 'after', 'op'\"\n                                        + \" and schema of fields 'before' and 'after' must match the table schema: %s\",\n                                IDENTIFIER, schemaString));\n            }\n        }\n    }\n}\n","sourceCodeStart":172,"sourceCodeEnd":199,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java#L172-L199","documentation":"Thrown by DebeziumAvroFormatFactory.validateSchemaString() during encoding-format creation when the user-supplied 'debezium-avro-confluent.schema' option, converted to a LogicalType via AvroSchemaConverter, does not equal the table's row type. The schema must be the exact Debezium envelope: a nullable record with fields 'before', 'after', 'op', where before/after match the table columns field-for-field and type-for-type.","triggerScenarios":"Passing a schema string whose before/after records differ from the sink table's column names, order, nullability, or types; passing a business-record schema instead of the Debezium envelope; missing or misspelled 'op' field; field order differences between the supplied Avro schema and the DDL.","commonSituations":"Hand-writing the schema option instead of copying the schema Debezium registered; schema drift after adding a column to the DDL without updating the schema option; JSON-vs-Avro name normalization differences (e.g. case-sensitive field names).","solutions":["Fetch the exact envelope schema from the registry (subject used by Debezium) and paste that JSON into the 'schema' option, or drop the option if you can rely on table-schema-derived envelope.","Align the sink table DDL columns (names, order, types, nullability) with the before/after records in the supplied schema.","Re-check that the schema includes top-level fields before, after, op and that before/after have identical payloads equal to the table schema."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Pre-check in a unit test before the job:\n// AvroSchemaConverter.convertToDataType(schemaJson).getLogicalType()\n//   .equals(tableRowType) — run the same equality the factory runs.\n// Best: derive the schema from the table and skip the 'schema' option entirely.","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Prefer omitting the 'schema' option when the table schema is authoritative","When supplying it, fetch the JSON verbatim from the registry subject Debezium writes","Diff before/after in CI whenever DDL columns change"],"tags":["debezium","avro","schema-validation","format-options"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}