apache/flink · error · IllegalArgumentException

Schema provided for '%s' format must be a nullable record ty

Error message

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

What it means

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.

Source

Thrown at flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java:190

        options.add(SCHEMA);
        options.add(SSL_KEYSTORE_LOCATION);
        options.add(SSL_KEYSTORE_PASSWORD);
        options.add(SSL_TRUSTSTORE_LOCATION);
        options.add(SSL_TRUSTSTORE_PASSWORD);
        options.add(BASIC_AUTH_CREDENTIALS_SOURCE);
        options.add(BASIC_AUTH_USER_INFO);
        options.add(BEARER_AUTH_CREDENTIALS_SOURCE);
        options.add(BEARER_AUTH_TOKEN);
        return options;
    }

    static void validateSchemaString(@Nullable String schemaString, RowType rowType) {
        if (schemaString != null) {
            LogicalType convertedDataType =
                    AvroSchemaConverter.convertToDataType(schemaString).getLogicalType();

            if (!convertedDataType.equals(rowType)) {
                throw new IllegalArgumentException(
                        format(
                                "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",
                                IDENTIFIER, schemaString));
            }
        }
    }
}

View on GitHub (pinned to 2f3c205e92)

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.

Defensive patterns

Strategy: validation

Validate before calling

// Pre-check in a unit test before the job:
// AvroSchemaConverter.convertToDataType(schemaJson).getLogicalType()
//   .equals(tableRowType) — run the same equality the factory runs.
// Best: derive the schema from the table and skip the 'schema' option entirely.

Prevention

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

When it happens

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

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

Related errors

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