Copilot commented on code in PR #4523:
URL: https://github.com/apache/flink-cdc/pull/4523#discussion_r4267525254
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java:
##########
@@ -157,14 +165,35 @@ protected void onSplitFinished(Map finishedSplitIds) {
for (Object splitState : finishedSplitIds.values()) {
SourceSplitBase sourceSplit = ((SourceSplitState)
splitState).toSourceSplit();
if (sourceSplit.isStreamSplit()) {
- StreamSplit streamSplit = sourceSplit.asStreamSplit();
+ // A stream split finishes for one of two reasons, see
+ // IncrementalSourceReader#onSplitFinished: it was suspended
by the enumerator so
+ // that newly added tables can be snapshotted, or it reached
its ending offset.
+ // Only the latter means the bounded snapshot-only read is
over and the slot is no
+ // longer needed.
+ //
+ // Comparing the split's starting and ending offsets cannot
answer that question.
+ // For snapshot-only, HybridSplitAssigner#createStreamSplit
seeds them with the
+ // lowest and highest high watermark of the finished snapshot
splits, and afterwards
+ // only IncrementalSourceRecordEmitter#updateStreamSplitState
advances the starting
+ // offset, for data-change records and heartbeats. A split
that reaches its ending
+ // offset without emitting such a record — a captured
publication with no traffic
+ // emits none — therefore left the slot behind on every run.
if (this.sourceConfig.getStartupOptions().isSnapshotOnly()
- && streamSplit
- .getStartingOffset()
- .isAtOrAfter(streamSplit.getEndingOffset())) {
+ &&
!incrementalSourceReaderContext.isStreamSplitReaderSuspended()) {
PostgresDialect dialect = (PostgresDialect) this.dialect;
- boolean removed =
dialect.removeSlot(dialect.getSlotName());
- LOG.info("Remove slot '{}' result is {}.",
dialect.getSlotName(), removed);
+ String slotName = dialect.getSlotName();
+ boolean removed = dialect.removeSlot(slotName);
+ if (removed) {
+ LOG.info("Removed replication slot '{}'.", slotName);
+ } else {
+ // removeSlot swallows the failure, so without this
the only trace of a
+ // slot that keeps pinning WAL would be an INFO line
reading "false".
+ LOG.warn(
+ "Failed to remove replication slot '{}'. It
will keep retaining WAL "
+ + "until it is dropped, e.g. with
pg_drop_replication_slot('{}').",
+ slotName,
+ slotName);
+ }
Review Comment:
Dropping the replication slot from every reader that observes the stream
split finishing can produce noisy WARN logs or misleading ‘failure’ messages in
parallel execution (e.g., only one subtask can drop the shared slot; others
will see `removed=false`). Consider scoping slot removal to a single
deterministic place (e.g., only subtask 0 via reader context index, or
coordinator/enumerator side), or at least treating ‘slot already absent’ as a
benign outcome to avoid false alarms.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java:
##########
@@ -189,6 +195,95 @@ void testNotifyCheckpointWindowSizeDefault() throws
Exception {
assertThat(completedCheckpointIds).containsExactly(101L);
}
+ /**
+ * A snapshot-only stream split that reached its ending offset must
release the replication slot
+ * even though its starting offset is behind that ending offset.
+ *
+ * <p>That combination is the normal case rather than an edge case: {@code
+ * HybridSplitAssigner#createStreamSplit} seeds the two offsets with the
lowest and highest high
+ * watermark of the finished snapshot splits, so they differ as soon as
there is more than one
+ * split or any concurrent write, and only emitted data-change or
heartbeat records move the
+ * starting offset afterwards.
+ */
+ @Test
+ void testSnapshotOnlyRemovesSlotWhenStartingOffsetIsBehindEndingOffset()
throws Exception {
+ final PostgresSourceReader reader = createSnapshotOnlyReader();
+ createSlot();
+ assertThat(customDatabase.checkSlot(slotName)).isEqualTo(slotName);
+
+ reader.onSplitFinished(finishedStreamSplit());
+
+ // checkSlot returns a human-readable "does not exist" sentence that
itself embeds the slot
+ // name, so compare for equality rather than containment.
+ assertThat(customDatabase.checkSlot(slotName)).isNotEqualTo(slotName);
+ }
+
+ /**
+ * A suspended stream split is the other reason {@code onSplitFinished}
fires — the enumerator
+ * paused streaming so newly added tables can be snapshotted, and the
split will resume. The
+ * slot has to survive that, otherwise resuming loses its position.
+ */
+ @Test
+ void testSnapshotOnlyKeepsSlotWhenStreamSplitIsMerelySuspended() throws
Exception {
+ final PostgresSourceReader reader = createSnapshotOnlyReader();
+ createSlot();
+ // StreamSplitUpdateRequestEvent is what the enumerator sends to pause
streaming; note the
+ // reader context flag is the authority here, not
StreamSplit#isSuspended.
+ reader.handleSourceEvents(new StreamSplitUpdateRequestEvent());
+
+ reader.onSplitFinished(finishedStreamSplit());
+
+ assertThat(customDatabase.checkSlot(slotName)).isEqualTo(slotName);
+ }
+
+ private Map<String, SourceSplitState> finishedStreamSplit() {
+ // Deliberately startingOffset < endingOffset: that is what the
assigner produces for a
+ // multi-split snapshot, and what the previous guard treated as "not
finished".
+ final Offset startingOffset = lsnOffset(1000L);
+ final Offset endingOffset = lsnOffset(2000L);
+ final StreamSplit split =
+ new StreamSplit(
+ StreamSplit.STREAM_SPLIT_ID,
+ startingOffset,
+ endingOffset,
+ new ArrayList<>(),
+ new HashMap<>(),
+ 0,
+ false,
+ true);
+ final Map<String, SourceSplitState> finished = new HashMap<>();
+ finished.put(split.splitId(), new StreamSplitState(split));
+ return finished;
+ }
+
+ private static Offset lsnOffset(long lsn) {
+ final Map<String, String> offset = new HashMap<>();
+ offset.put("lsn", Long.toString(lsn));
+ return PostgresOffset.of(offset);
+ }
+
+ private void createSlot() throws Exception {
+ try (Connection connection =
+ getJdbcConnection(POSTGRES_CONTAINER,
customDatabase.getDatabaseName());
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "SELECT pg_create_logical_replication_slot('%s',
'pgoutput')",
+ slotName));
+ }
Review Comment:
This test constructs SQL by string-formatting `slotName` directly into the
query. Even in tests, this pattern is brittle (breaks on quotes/special chars)
and can normalize unsafe practices. Prefer using a PreparedStatement (bind
`slotName`) or at least escaping single quotes in `slotName` before
interpolation.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java:
##########
@@ -1363,6 +1410,23 @@ private static long getCountOfTable(JdbcConnection jdbc,
TableId tableId) throws
});
}
+ private long getSlotCount(JdbcConnection jdbc) throws SQLException {
+ final String query =
+ String.format(
+ "SELECT count(*) FROM pg_replication_slots WHERE
slot_name = '%s'",
+ slotName);
+ return jdbc.queryAndMap(
+ query,
+ rs -> {
+ if (!rs.next()) {
+ throw new SQLException(
+ String.format(
+ "No result returned after running
query [%s]", query));
+ }
+ return rs.getLong(1);
+ });
+ }
Review Comment:
Like `createSlot()`, this query embeds `slotName` directly into SQL. Using
parameter binding (if supported by `JdbcConnection` helpers) or escaping would
make the test more robust (e.g., if `slotName` contains quotes) and avoids
propagating unsafe SQL construction patterns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
