{"record":{"id":"2e481c4024ce774b","repo":"apache/seatunnel","slug":"read-committed-offset-failed","errorCode":"READ_COMMITTED_OFFSET_FAILED","errorMessage":"Could not read PostgreSQL replication slot '${slotName}'. Verify that the slot still exists and is not being dropped.","messagePattern":"Could not read PostgreSQL replication slot '(.+?)'\\. Verify that the slot still exists and is not being dropped\\.","errorType":"error_code","errorClass":"SeaTunnelRuntimeException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactory.java","lineNumber":90,"sourceCode":"\n    @Override\n    public Offset committedOffset() {\n        String slotName = sourceConfig.getDbzConfiguration().getString(\"slot.name\");\n        try (JdbcConnection jdbcConnection = dialect.openJdbcConnection(sourceConfig)) {\n            try (PreparedStatement statement =\n                    jdbcConnection\n                            .connection()\n                            .prepareStatement(\n                                    \"SELECT confirmed_flush_lsn::text, active_pid FROM pg_replication_slots WHERE slot_name = ?\")) {\n                statement.setString(1, slotName);\n                try (ResultSet resultSet = statement.executeQuery()) {\n                    return readCommittedOffset(resultSet, slotName);\n                }\n            }\n        } catch (SeaTunnelRuntimeException e) {\n            throw e;\n        } catch (Exception e) {\n            throw new SeaTunnelRuntimeException(\n                    PostgresConnectorErrorCode.READ_COMMITTED_OFFSET_FAILED,\n                    String.format(\n                            \"Could not read PostgreSQL replication slot '%s'. Verify that the slot still exists and is not being dropped.\",\n                            slotName),\n                    e);\n        }\n    }\n\n    static LsnOffset readCommittedOffset(ResultSet resultSet, String slotName) throws SQLException {\n        if (!resultSet.next()) {\n            throw invalidReplicationSlot(\n                    slotName, \"does not exist; create the logical replication slot first\");\n        }\n        Object activePid = resultSet.getObject(2);\n        if (activePid != null) {\n            throw invalidReplicationSlot(\n                    slotName,\n                    String.format(","sourceCodeStart":72,"sourceCodeEnd":108,"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/offset/LsnOffsetFactory.java#L72-L108","documentation":"LsnOffsetFactory.committedOffset() queries pg_replication_slots for the confirmed_flush_lsn of the configured slot and throws a SeaTunnelRuntimeException with code READ_COMMITTED_OFFSET_FAILED when the lookup fails unexpectedly (SQL error, connection drop, etc.). It tells the user to verify the replication slot still exists, since a missing or dropped slot breaks offset recovery.","triggerScenarios":"Reading the committed offset at startup/restart when: the slot named by `slot.name` was dropped (manually or by the DB due to `max_slot_wal_keep_size`), the query on pg_replication_slots throws, or the JDBC connection fails mid-query. A plain SeaTunnelRuntimeException is rethrown unchanged; all other exceptions are wrapped.","commonSituations":"Slots dropped by DBAs to free WAL disk space; slot removed because the job was down past wal_keep_size; typo in slot.name; insufficient privilege to read pg_replication_slots.","solutions":["Check `SELECT slot_name, active, confirmed_flush_lsn FROM pg_replication_slots;` on the database to confirm the slot exists.","Recreate the slot or start a new job with a fresh slot name if it was dropped (data since the last LSN is lost).","Fix `slot.name` in the source config if it is misspelled.","Grant the replication user permission to query pg_replication_slots and resolve any underlying connection errors from the cause."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Run before job start\nSELECT slot_name, active, confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = '<slot.name>';","typeGuard":null,"tryCatchPattern":"try {\n    offsetFactory.committedOffset();\n} catch (SeaTunnelRuntimeException e) {\n    if (e.getErrorCode() == PostgresConnectorErrorCode.READ_COMMITTED_OFFSET_FAILED) {\n        // recreate slot or fail fast with clear ops message\n    }\n}","preventionTips":["Monitor pg_replication_slots and alert on missing/inactive slots","Give each job a unique, stable slot.name","Protect slots from wal_keep_size/max_slot_wal_keep_size cleanup","Grant the replication user read access to pg_replication_slots"],"tags":["postgresql","cdc","replication-slot","offset","checkpoint"],"backgroundTag":"resource-not-found","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"}