tchivs commented on code in PR #4523:
URL: https://github.com/apache/flink-cdc/pull/4523#discussion_r4278078423
##########
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:
slotName is a fixed test constant, so string formatting is fine here.
##########
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:
slotName is a fixed test constant, so string formatting is fine here.
--
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]
