apache/flink · error · ValidationException

Option %s.%s is required for serialization

Error message

Option %s.%s is required for serialization

What it means

ValidationException from RegistryAvroFormatFactory.createEncodingFormat: the 'avro-confluent-registry.subject' option is mandatory for sinks. Serialization must know under which subject to register the schema; unlike deserialization it cannot be derived from the data.

Source

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

            @Override
            public ChangelogMode getChangelogMode() {
                return ChangelogMode.insertOnly();
            }
        };
    }

    @Override
    public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
            DynamicTableFactory.Context context, ReadableConfig formatOptions) {
        FactoryUtil.validateFactoryOptions(this, formatOptions);

        String schemaRegistryURL = formatOptions.get(URL);
        Optional<String> subject = formatOptions.getOptional(SUBJECT);
        Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
        Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);

        if (!subject.isPresent()) {
            throw new ValidationException(
                    String.format(
                            "Option %s.%s is required for serialization",
                            IDENTIFIER, SUBJECT.key()));
        }

        return new EncodingFormat<SerializationSchema<RowData>>() {
            @Override
            public SerializationSchema<RowData> createRuntimeEncoder(
                    DynamicTableSink.Context context, DataType consumedDataType) {
                final RowType rowType = (RowType) consumedDataType.getLogicalType();
                final Schema schema =
                        schemaString
                                .map(s -> getAvroSchema(s, rowType))
                                .orElse(AvroSchemaConverter.convertToSchema(rowType));
                return new AvroRowDataSerializationSchema(
                        rowType,
                        ConfluentRegistryAvroSerializationSchema.forGeneric(
                                subject.get(), schema, schemaRegistryURL, optionalPropertiesMap),

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Add 'avro-confluent-registry.subject' = '<subject-name>' (typically '<topic>-value') to the WITH clause.
2. Optionally also set 'avro-confluent-registry.schema' if you must pin an explicit schema instead of deriving it from the table.

Example fix

-- before
WITH (
  'connector'='kafka', 'topic'='t', 'format'='avro-confluent-registry',
  'avro-confluent-registry.schema-registry.url'='http://reg:8081'
)

-- after
WITH (
  'connector'='kafka', 'topic'='t', 'format'='avro-confluent-registry',
  'avro-confluent-registry.schema-registry.url'='http://reg:8081',
  'avro-confluent-registry.subject'='t-value'
)

Map<String,String> opts = tableOptions; // your WITH options
boolean isSink = /* connector is a sink */;
if (isSink && "avro-confluent-registry".equals(opts.get("format"))
        && !opts.containsKey("avro-confluent-registry.subject")) {
    throw new ValidationException(
        "Sinks with format 'avro-confluent-registry' require 'avro-confluent-registry.subject'");
}

try {
    tableEnv.executeSql(createTableSql);
} catch (ValidationException e) {
    if (e.getMessage().contains("is required for serialization")) {
        // add 'avro-confluent-registry.subject' = '<topic>-value' to WITH clause
    }
    throw e;
}

When it happens

Trigger: Creating a table with format 'avro-confluent-registry' used as a SINK without 'avro-confluent-registry.subject' in the WITH options; connector options copied from a source table that never needed subject.

Common situations: Converting a Kafka source DDL into a sink DDL and forgetting to add the subject; sink tables where the user assumed the topic name implies the subject.

Related errors

• Failed to serialize schema registry.
• Unknown data format. Magic number does not match
• Could not find schema with id %s in registry
• Could not register schema in registry
• Schema provided for '%s' format does not match the table sch

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