{"record":{"id":"0dc897571dc04dc9","repo":"apache/seatunnel","slug":"the-db-connection-is-not-a-valid-replication-conne","errorCode":null,"errorMessage":"The DB connection is not a valid replication connection","messagePattern":"The DB connection is not a valid replication connection","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java","lineNumber":330,"sourceCode":"            // there's no info for this plugin and slot so create a new slot\n            if (shouldCreateSlot) {\n                this.createReplicationSlot();\n            }\n\n            // replication connection does not support parsing of SQL statements so we need to\n            // create\n            // the connection without executing on connect statements - see JDBC opt\n            // preferQueryMode=simple\n            pgConnection();\n            final String identifySystemStatement = \"IDENTIFY_SYSTEM\";\n            LOGGER.debug(\n                    \"running '{}' to validate replication connection\", identifySystemStatement);\n            final Lsn xlogStart =\n                    queryAndMap(\n                            identifySystemStatement,\n                            rs -> {\n                                if (!rs.next()) {\n                                    throw new IllegalStateException(\n                                            \"The DB connection is not a valid replication connection\");\n                                }\n                                String xlogpos = rs.getString(\"xlogpos\");\n                                LOGGER.debug(\"received latest xlogpos '{}'\", xlogpos);\n                                return Lsn.valueOf(xlogpos);\n                            });\n\n            if (slotCreationInfo != null) {\n                this.defaultStartingPos = slotCreationInfo.startLsn();\n            } else if (shouldCreateSlot || !slotInfo.hasValidFlushedLsn()) {\n                // this is a new slot or we weren't able to read a valid flush LSN pos, so we always\n                // start from the xlog pos that was reported\n                this.defaultStartingPos = xlogStart;\n            } else {\n                Lsn latestFlushedLsn = slotInfo.latestFlushedLsn();\n                this.defaultStartingPos =\n                        latestFlushedLsn.compareTo(xlogStart) < 0 ? latestFlushedLsn : xlogStart;\n                if (LOGGER.isDebugEnabled()) {","sourceCodeStart":312,"sourceCodeEnd":348,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java#L312-L348","documentation":"Raised during initReplicationSlot after executing IDENTIFY_SYSTEM on the replication connection. IDENTIFY_SYSTEM must return exactly one row containing the current xlogpos; an empty result set means the JDBC connection, although opened with replication=database, is not actually a valid physical replication connection. The library throws IllegalStateException because this is an internal invariant violation.","triggerScenarios":"The connection URL used for streaming does not include replication=database (or an equivalent replication mode), the driver silently downgraded the connection, or the server accepted the connection but the IDENTIFY_SYSTEM result set came back empty (non-Postgres/openGauss compatible endpoint).","commonSituations":"Misconfigured JDBC URL parameters in connector config (missing replication=true/database param); connecting through a proxy or pooler (e.g. PgBouncer in transaction mode) that strips replication capabilities; pointing the connector at a non-PostgreSQL-compatible database that pretends to speak the wire protocol; driver version mismatch overriding replication mode.","solutions":["Check the replication connection URL includes the replication=database parameter and the correct user with REPLICATION privilege.","Bypass connection poolers/proxies (PgBouncer transaction pooling) and connect directly to the primary server for the streaming connection.","Verify the target is a genuine PostgreSQL/openGauss primary: run `psql \"replication=database\" -c IDENTIFY_SYSTEM` manually with the same credentials.","Confirm the PostgreSQL JDBC driver version matches what the connector expects; upgrade the connector/driver if the replication parameter is being dropped.","Ensure you connect to the primary, not a hot standby without appropriate settings, and that wal_level=logical where required."],"exampleFix":"// before (misconfigured URL, no replication mode)\nString url = \"jdbc:postgresql://host:5432/mydb\";\n// after\nString url = \"jdbc:postgresql://host:5432/mydb?replication=database\";","handlingStrategy":"validation","validationCode":"// validate replication capability before starting the connector\ntry (Connection c = DriverManager.getConnection(\n        \"jdbc:postgresql://host:5432/mydb?replication=database\", user, pass);\n     Statement s = c.createStatement();\n     ResultSet rs = s.executeQuery(\"IDENTIFY_SYSTEM\")) {\n    if (!rs.next()) throw new IllegalStateException(\"endpoint is not a valid replication connection\");\n    System.out.println(\"xlogpos=\" + rs.getString(\"xlogpos\"));\n}","typeGuard":null,"tryCatchPattern":"try {\n    engine.start();\n} catch (IllegalStateException e) {\n    if (e.getMessage().contains(\"not a valid replication connection\")) {\n        // fix JDBC URL / bypass pooler / check REPLICATION privilege, then retry\n    } else {\n        throw e;\n    }\n}","preventionTips":["Always include replication=database in the streaming connection URL.","Never route replication connections through PgBouncer or transaction-mode poolers.","Grant REPLICATION (or rds_replication on RDS) to the connector user.","Smoke-test with `psql \"replication=database\" -c IDENTIFY_SYSTEM` before deploying."],"tags":["postgresql","cdc","replication","jdbc","connection"],"backgroundTag":"empty-result-set","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"}