apache/seatunnel · error · ConnectException

Unable to filtered publication for

Error message

Unable to %s filtered publication %s for %s

What it means

Thrown by PostgresReplicationConnection when the CREATE or ALTER PUBLICATION SQL statement fails during filtered publication setup. The library wraps the underlying SQLException from conn.execute() in a ConnectException, indicating whether it was attempting to create a new publication or update an existing one. The publication name and table filter string are included so the operator can inspect the offending publication in the database.

Solutions

  1. Grant the connector user the required privileges (CREATE on the database, or make it owner of the publication, or a superuser/rds_replication role).
  2. Verify the table filter string in your config (table.include.list / schema include patterns) resolves to valid table names and the publication does not already exist with a conflicting definition.
  3. If the publication exists but is not owned by the connector user, either drop it (DROP PUBLICATION) so it can be recreated, or ALTER PUBLICATION <name> OWNER TO <connector_user>.
  4. Check the server logs / the wrapped cause (e) for the exact PostgreSQL error code and message; fix the SQL-level problem it reports.
  5. Set publication.autocreate.mode to 'disabled' and pre-create the publication manually if you cannot grant privileges.

Example fix

// before: connector fails creating filtered publication with limited user
GRANT CREATE ON DATABASE mydb TO cdc_user;
// or pre-create and transfer ownership:
CREATE PUBLICATION dbz_publication FOR TABLE public.orders;
ALTER PUBLICATION dbz_publication OWNER TO cdc_user;
Defensive patterns

Strategy: try-catch

Validate before calling

-- run before starting the connector
SELECT current_user, has_database_privilege(current_user, 'mydb', 'CREATE');
SELECT pubname, puballtables FROM pg_publication WHERE pubname = 'dbz_publication';
SELECT * FROM pg_roles WHERE rolname = 'cdc_user';

Try / catch

try {
    connector.start();
} catch (ConnectException e) {
    if (e.getMessage().startsWith("Unable to create filtered publication")) {
        // grant privileges or pre-create publication manually, then retry
    } else {
        throw e;
    }
}

Prevention

When it happens

Trigger: Executing CREATE PUBLICATION <name> FOR TABLE <filter> or ALTER PUBLICATION <name> SET TABLE <filter> (via initPublication -> createOrUpdatePublicationModeFilterted) fails: e.g. insufficient privileges, the publication does not exist when updating, or an invalid table filter expression.

Common situations: The DB user lacks CREATE or ownership rights on the database/publication; publication.autocreate.mode is 'filtered' but the configured table list has syntax errors; the publication was dropped manually while the connector expects to update it; running against openGauss/Postgres versions where ALTER PUBLICATION ... SET TABLE is unsupported; max publication limits reached (e.g. max_replication_slots or publication-per-database limits on managed services).

Understand the failure class

Background: "query failed", "%w: SQL error" — wrapped database query errors in Go libraries explained — this error's family across 3 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/087c384f73cd64c0. Report an issue: GitHub.

Appendix: source

Thrown at seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java:278

                                "No table filters found for filtered publication %s",
                                publicationName));
            }
            createOrUpdatePublicationStmt =
                    isUpdate
                            ? String.format(
                                    "ALTER PUBLICATION %s SET TABLE %s;",
                                    publicationName, tableFilterString)
                            : String.format(
                                    "CREATE PUBLICATION %s FOR TABLE %s;",
                                    publicationName, tableFilterString);
            LOGGER.info(
                    isUpdate
                            ? "Updating Publication with statement '{}'"
                            : "Creating Publication with statement '{}'",
                    createOrUpdatePublicationStmt);
            conn.execute(createOrUpdatePublicationStmt);
        } catch (Exception e) {
            throw new ConnectException(
                    String.format(
                            "Unable to %s filtered publication %s for %s",
                            isUpdate ? "update" : "create", publicationName, tableFilterString),
                    e);
        }
    }

    private Set<TableId> determineCapturedTables() throws Exception {
        Set<TableId> allTableIds = jdbcConnection.getAllTableIds(connectorConfig.databaseName());

        Set<TableId> capturedTables = new HashSet<>();

        for (TableId tableId : allTableIds) {
            if (tableFilter.dataCollectionFilter().isIncluded(tableId)) {
                LOGGER.trace("Adding table {} to the list of captured tables", tableId);
                capturedTables.add(tableId);
            } else {
                LOGGER.trace(

View on GitHub (pinned to cf67b549a7)