{"record":{"id":"9f372227c0b5ad00","repo":"apache/flink","slug":"debezium-json-serialization-doesn-t-support-s-s","errorCode":null,"errorMessage":"Debezium JSON serialization doesn't support '%s.%s' option been set to true.","messagePattern":"Debezium JSON serialization doesn't support '(.+?)\\.(.+?)' option been set to true\\.","errorType":"validation","errorClass":"ValidationException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonFormatFactory.java","lineNumber":159,"sourceCode":"        options.add(JSON_MAP_NULL_KEY_MODE);\n        options.add(JSON_MAP_NULL_KEY_LITERAL);\n        options.add(ENCODE_DECIMAL_AS_PLAIN_NUMBER);\n        options.add(ENCODE_IGNORE_NULL_FIELDS);\n        return options;\n    }\n\n    /** Validator for debezium decoding format. */\n    private static void validateDecodingFormatOptions(ReadableConfig tableOptions) {\n        JsonFormatOptionsUtil.validateDecodingFormatOptions(tableOptions);\n    }\n\n    /** Validator for debezium encoding format. */\n    private static void validateEncodingFormatOptions(ReadableConfig tableOptions) {\n        JsonFormatOptionsUtil.validateEncodingFormatOptions(tableOptions);\n\n        // validator for {@link SCHEMA_INCLUDE}\n        if (tableOptions.get(SCHEMA_INCLUDE)) {\n            throw new ValidationException(\n                    String.format(\n                            \"Debezium JSON serialization doesn't support '%s.%s' option been set to true.\",\n                            IDENTIFIER, SCHEMA_INCLUDE.key()));\n        }\n    }\n}\n","sourceCodeStart":141,"sourceCodeEnd":166,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonFormatFactory.java#L141-L166","documentation":"Thrown by the Debezium JSON encoding-format factory during table-environment validation. Debezium JSON serialization only writes the 'before', 'after', and 'op' envelope fields, so emitting an embedded 'schema' (JSON schema include) is not implemented. Setting 'debezium-json.schema-include' = true on a table used as a SINK therefore fails fast at DDL/CREATE time with a ValidationException.","triggerScenarios":"Creating a table with 'format' = 'debezium-json' and a connector that sinks (writes) to it, plus FORMAT option 'debezium-json.schema-include' = true. The factory method validateEncodingFormatOptions checks tableOptions.get(SCHEMA_INCLUDE) and throws before any job runs.","commonSituations":"Copy-pasting a Debezium source (kafka connector reading Debezium topics with schema-include true) table DDL and reusing it as a sink target; assuming the decoding option is symmetric for encoding; migrating from JSON format where json schema-include is tolerated.","solutions":["Remove 'debezium-json.schema-include' = true (or set it to false) from the sink table's WITH options.","If you meant to consume (read) Debezium messages that carry a schema, use the table as a SOURCE instead — schema-include is only supported for decoding.","If you need to emit schemas with CDC envelopes, use a different CDC format (e.g. Canal JSON with its own options, or Debezium Avro via confluent-registry) that supports schema emission."],"exampleFix":"-- before\nCREATE TABLE kafka_sink (...) WITH (\n  'connector' = 'kafka',\n  'topic' = 'cdc-out',\n  'format' = 'debezium-json',\n  'debezium-json.schema-include' = 'true'\n);\n-- after\nCREATE TABLE kafka_sink (...) WITH (\n  'connector' = 'kafka',\n  'topic' = 'cdc-out',\n  'format' = 'debezium-json'\n);","handlingStrategy":"validation","validationCode":"// Before creating the sink table, check the format options\nboolean schemaInclude = tableOptions.getOptional(JsonFormatOptionsUtil.JSON_SCHEMA_INCLUDE)\n        .orElse(false);\nboolean isSink = /* table is written to */;\nif (isSink && \"debezium-json\".equals(format) && schemaInclude) {\n    throw new IllegalArgumentException(\n        \"Remove 'debezium-json.schema-include' for debezium-json sinks\");\n}","typeGuard":null,"tryCatchPattern":"No catch needed: it fails at DDL validation time. Wrap Catalog/TableEnvironment executeSql in try-catch ValidationException to surface a friendly message if you build DDL dynamically.","preventionTips":["Keep source-side and sink-side CDC format option sets in separate DDL templates.","Lint generated DDL: reject 'schema-include' = true whenever the table is a sink."],"tags":["flink","debezium","json","cdc","table-api","validation"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}