[source-postgres] Reversed LSN arguments in position(SourceRecord) affect CDC progress comparisons
- Linguagem predominante
- Python
- Estrelas
- 22.1k
- Forks
- 5.4k
- Merge médio
- 5h
- PRs com merge (30d)
- 671
Descrição
### Connector Name
source-postgres
### Connector Version
3.8.1 and 3.8.5 (official connector class files inspected).
### What step the error happened?
During the sync
### Relevant information
`PostgresSourceDebeziumOperations.position(SourceRecord)` passes two LSN values to `PostgresSourceCdcPosition` in the opposite order from its constructor. This can change the ordering of comparison objects when the values differ.
The constructor takes `lsnCommit` first and `lsn` second. The SourceRecord overload instead supplies the offset's `lsn` first:
```kotlin
val lsn = sourceRecord.sourceOffset()[LSN] as Long?
val lsnCommit = sourceRecord.sourceOffset()[LSN_COMMIT] as Long?
return PostgresSourceCdcPosition(Lsn.valueOf(lsn), Lsn.valueOf(lsnCommit))
```
Both arguments have the same type, so the positional call compiles. The comparator prioritizes the field named `lsnCommit`, using `lsn` only to break ties. Other position-building overloads use named arguments and preserve the intended mapping.
Public source references at commit `b6bfc635ffb82f71ae178d66b7b0a81b1895927c`:
- [Constructor and comparator](https://github.com/airbytehq/airbyte/blob/b6bfc635ffb82f71ae178d66b7b0a81b1895927c/airbyte-integrations/connectors/source-postgres/src/main/kotlin/io/airbyte/integrations/source/postgres/cdc/PostgresSourceCdcPosition.kt#L10-L28).
- [SourceRecord overload](https://github.com/airbytehq/airbyte/blob/b6bfc635ffb82f71ae178d66b7b0a81b1895927c/airbyte-integrations/connectors/source-postgres/src/main/kotlin/io/airbyte/integrations/source/postgres/cdc/PostgresSourceDebeziumOperations.kt#L399-L403).
**Synthetic mapping and comparison case**
For a SourceRecord with `sourceOffset()` containing `lsn=100L` and `lsn_commit=200L`:
- Expected position fields: `lsnCommit=200`, `lsn=100`.
- Actual assignment in the inspected method: `lsnCommit=100`, `lsn=200`.
These illustrative offsets demonstrate the resulting comparison error. All numbers are synthetic:
| Input | Offset lsn | Offset lsn_commit |
| --- | ---: | ---: |
| Previous | 190 | 200 |
| Current | 110 | 300 |
With correct field assignment, current sorts after previous because 300 is greater than 200. With the reversed assignment, current sorts before previous because the comparator's first field contains 110 versus 190.
This demonstrates the mapping and ordering consequence. It is not a reproduction of a complete database transaction sequence or connector sync.
**Compiled-code verification and relevant caller**
Disassembly with `javap -p -c` confirmed the reversed assignment in the inspected 3.8.1 and 3.8.5 connector classes. The SourceRecord overloads have identical instructions after normalizing constant-pool reference numbers. The position class containing the constructor and comparator is byte-for-byte identical in those two inspected versions.
In the inspected 3.8.1 CDC reader, `CdcPartitionReader.EventConsumer.findCloseReason` prefers `position(SourceRecord)` when available. It uses that position for heartbeat progress and target-bound comparisons. Consequently, the reversed assignment affects inputs to those decisions.
The affected line builds a comparison object; it does not directly rewrite the serialized checkpoint. This report does not claim that the defect alone causes a particular timeout, checkpoint failure, or data loss.
**Suggested correction and coverage**
Use named constructor arguments so the offset's `lsn_commit` populates `lsnCommit` and its `lsn` populates `lsn`. Add regression coverage for unequal values and for an offset whose commit position increases while its row position decreases. Also verify consistency between the SourceRecord and other position-building overloads.
### Relevant log output
No production logs or environment details are included. This report is based on source and compiled-code inspection, with synthetic comparison inputs.
### Contribute
- [x] Yes, I want to contribute
Guia de contribuição
Direção de pesquisa
Start with PostgresSourceDebeziumOperations.kt at position(SourceRecord), then compare its constructor call with the named-argument overloads and the comparator in PostgresSourceCdcPosition.kt. Add regression coverage for unequal lsn and lsn_commit values, including an increasing commit position with a decreasing row position, and verify consistent SourceRecord and overload behavior.
Escrita pelo modelo de indexação a partir do texto da issue.
Avaliação
- Stack de tecnologia
- kotlin, postgresql
- Domínio
- data-engineering, databases
- Tipo de issue
- Bug
- Dificuldade
- 3/5
- Tempo estimado
- 1-2 dias
- Status de atividade
- Ativa
- Clareza
- Claramente especificada
- Facilidade para iniciantes
- 76/100