{"record":{"id":"c059ac9faf1183b2","repo":"apache/flink","slug":"option-s-s-is-required-for-serialization","errorCode":null,"errorMessage":"Option %s.%s is required for serialization","messagePattern":"Option (.+?)\\.(.+?) is required for serialization","errorType":"validation","errorClass":"ValidationException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java","lineNumber":134,"sourceCode":"            @Override\n            public ChangelogMode getChangelogMode() {\n                return ChangelogMode.insertOnly();\n            }\n        };\n    }\n\n    @Override\n    public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(\n            DynamicTableFactory.Context context, ReadableConfig formatOptions) {\n        FactoryUtil.validateFactoryOptions(this, formatOptions);\n\n        String schemaRegistryURL = formatOptions.get(URL);\n        Optional<String> subject = formatOptions.getOptional(SUBJECT);\n        Optional<String> schemaString = formatOptions.getOptional(SCHEMA);\n        Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);\n\n        if (!subject.isPresent()) {\n            throw new ValidationException(\n                    String.format(\n                            \"Option %s.%s is required for serialization\",\n                            IDENTIFIER, SUBJECT.key()));\n        }\n\n        return new EncodingFormat<SerializationSchema<RowData>>() {\n            @Override\n            public SerializationSchema<RowData> createRuntimeEncoder(\n                    DynamicTableSink.Context context, DataType consumedDataType) {\n                final RowType rowType = (RowType) consumedDataType.getLogicalType();\n                final Schema schema =\n                        schemaString\n                                .map(s -> getAvroSchema(s, rowType))\n                                .orElse(AvroSchemaConverter.convertToSchema(rowType));\n                return new AvroRowDataSerializationSchema(\n                        rowType,\n                        ConfluentRegistryAvroSerializationSchema.forGeneric(\n                                subject.get(), schema, schemaRegistryURL, optionalPropertiesMap),","sourceCodeStart":116,"sourceCodeEnd":152,"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#L116-L152","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Add 'avro-confluent-registry.subject' = '<subject-name>' (typically '<topic>-value') to the WITH clause.","Optionally also set 'avro-confluent-registry.schema' if you must pin an explicit schema instead of deriving it from the table."],"exampleFix":"-- before\nWITH (\n  'connector'='kafka', 'topic'='t', 'format'='avro-confluent-registry',\n  'avro-confluent-registry.schema-registry.url'='http://reg:8081'\n)\n\n-- after\nWITH (\n  'connector'='kafka', 'topic'='t', 'format'='avro-confluent-registry',\n  'avro-confluent-registry.schema-registry.url'='http://reg:8081',\n  'avro-confluent-registry.subject'='t-value'\n)","handlingStrategy":"validation","validationCode":"Map<String,String> opts = tableOptions; // your WITH options\nboolean isSink = /* connector is a sink */;\nif (isSink && \"avro-confluent-registry\".equals(opts.get(\"format\"))\n        && !opts.containsKey(\"avro-confluent-registry.subject\")) {\n    throw new ValidationException(\n        \"Sinks with format 'avro-confluent-registry' require 'avro-confluent-registry.subject'\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    tableEnv.executeSql(createTableSql);\n} catch (ValidationException e) {\n    if (e.getMessage().contains(\"is required for serialization\")) {\n        // add 'avro-confluent-registry.subject' = '<topic>-value' to WITH clause\n    }\n    throw e;\n}","preventionTips":["Template sink DDLs with the subject option pre-filled (convention: <topic>-value).","Run CREATE TABLE statements through a DDL linter that knows per-format required options."],"tags":["avro","confluent","schema-registry","validation","flink","config"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}