{"record":{"id":"da67dca45b7c3960","repo":"apache/seatunnel","slug":"postgresql-cdc-schema-evolution-requires-decoding","errorCode":null,"errorMessage":"PostgreSQL-CDC schema evolution requires 'decoding.plugin.name = pgoutput' because PostgreSQL RELATION messages provide the changed schema.","messagePattern":"PostgreSQL-CDC schema evolution requires 'decoding\\.plugin\\.name = pgoutput' because PostgreSQL RELATION messages provide the changed schema\\.","errorType":"validation","errorClass":"SeaTunnelException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java","lineNumber":184,"sourceCode":"                options.getOptional(PostgresIncrementalSourceOptions.SLOT_NAME)\n                        .map(String::trim)\n                        .filter(name -> !name.isEmpty());\n        if (!slotName.isPresent()) {\n            throw new SeaTunnelException(\n                    String.format(\n                            \"PostgreSQL-CDC startup.mode '%s' requires an explicit '%s' option.\",\n                            StartupMode.COMMITTED_OFFSET,\n                            PostgresIncrementalSourceOptions.SLOT_NAME.key()));\n        }\n    }\n\n    private void validateSchemaEvolutionOptions(ReadonlyConfig options) {\n        if (options.get(SourceOptions.SCHEMA_CHANGES_ENABLED)\n                && !\"pgoutput\"\n                        .equalsIgnoreCase(\n                                options.get(\n                                        PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME))) {\n            throw new SeaTunnelException(\n                    String.format(\n                            \"PostgreSQL-CDC schema evolution requires '%s = pgoutput' because PostgreSQL RELATION messages provide the changed schema.\",\n                            PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME.key()));\n        }\n    }\n\n    private Map<TableId, Struct> tableChanges() {\n        JdbcSourceConfig jdbcSourceConfig = configFactory.create(0);\n        PostgresDialect dialect =\n                new PostgresDialect(\n                        (PostgresSourceConfigFactory) configFactory,\n                        catalogTables,\n                        requireReplicaIdentityFull);\n        List<TableId> discoverTables = dialect.discoverDataCollections(jdbcSourceConfig);\n        SchemaNameAdjuster adjuster = SchemaNameAdjuster.create();\n        ConnectTableChangeSerializer connectTableChangeSerializer =\n                new ConnectTableChangeSerializer(adjuster);\n        try (JdbcConnection jdbcConnection = dialect.openJdbcConnection(jdbcSourceConfig)) {","sourceCodeStart":166,"sourceCodeEnd":202,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java#L166-L202","documentation":"SeaTunnel PostgreSQL CDC throws this during startup validation when schema evolution is enabled (`schema-changes.enabled = true`) but the logical decoding plugin is not `pgoutput`. Only pgoutput emits PostgreSQL RELATION messages carrying the changed table schema, which the schema-evolution machinery needs to resolve DDL changes; other plugins (e.g. decoderbufs, wal2json) cannot supply that information.","triggerScenarios":"Calling PostgresIncrementalSource builder with `.schemaChangesEnabled(true)` (or config `schema-changes.enabled=true`) while `decoding.plugin.name` is unset (defaults to decoderbufs) or set to anything other than `pgoutput` (case-insensitive check).","commonSituations":"Users copying older CDC configs that used decoderbufs; enabling schema evolution on clusters where pgoutput is the default but the option was explicitly overridden; forgetting that slot creation with non-pgoutput plugins cannot convey schema.","solutions":["Set `decoding.plugin.name = pgoutput` in the source options.","Or disable schema evolution (`schema-changes.enabled = false`) if you do not need DDL tracking.","Ensure the replication user has sufficient privileges (pg_monitor / replication role) since pgoutput requires a logical replication slot."],"exampleFix":"// before\nPostgresIncrementalSource.<String>builder()\n    .schemaChangesEnabled(true)\n    .decodingPluginName(\"decoderbufs\")\n    .build();\n// after\nPostgresIncrementalSource.<String>builder()\n    .schemaChangesEnabled(true)\n    .decodingPluginName(\"pgoutput\")\n    .build();","handlingStrategy":"validation","validationCode":"if (schemaChangesEnabled && !\"pgoutput\".equalsIgnoreCase(decodingPluginName)) {\n    throw new IllegalArgumentException(\"schema-changes.enabled=true requires decoding.plugin.name=pgoutput\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always pair schema-changes.enabled=true with decoding.plugin.name=pgoutput","Centralize CDC source config building in one helper that enforces this invariant","Document the pairing requirement in your job templates"],"tags":["postgresql","cdc","config-validation","schema-evolution"],"backgroundTag":"invalid-config-value","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}