You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Mrart (via GitHub)" <gi...@apache.org> on 2025/09/03 12:56:05 UTC
[PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Mrart opened a new pull request, #4113:
URL: https://github.com/apache/flink-cdc/pull/4113
[Flink-38277] [postgres]drop slot when startup.mode=snapshot
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113#discussion_r2320705697
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java:
##########
@@ -148,8 +148,11 @@ public void configure(SourceSplitBase sourceSplitBase) {
dbzConfig
.getConfig()
.edit()
- // never drop slot for stream split, which is also global split
- .with(DROP_SLOT_ON_STOP.name(), false)
+ // When startup.mode='snapshot' drop slot for stream split,
+ // which is also global split
Review Comment:
I don't think so, If the job success to finish the stream split, it does no need to restart. However, If fails before complete stream split, it means that the job is not finished, and need to restart with checkpoint .
For batch mode, it has no checkpoint, thus no need to restart with checkpont. But streaming job, it do need. @LYanquan , WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113#discussion_r2320626321
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java:
##########
@@ -148,8 +148,11 @@ public void configure(SourceSplitBase sourceSplitBase) {
dbzConfig
.getConfig()
.edit()
- // never drop slot for stream split, which is also global split
- .with(DROP_SLOT_ON_STOP.name(), false)
+ // When startup.mode='snapshot' drop slot for stream split,
+ // which is also global split
Review Comment:
What if a exception is thrown, then job fails. If drop in this case, the job cannot restart! I do think we should do it in org.apache.flink.connector.base.source.reader.SourceReaderBase#onSplitFinished. We should determine whether it's is finished: stream split and endingOffset >= startingOffset
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on code in PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113#discussion_r2320673516
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java:
##########
@@ -148,8 +148,11 @@ public void configure(SourceSplitBase sourceSplitBase) {
dbzConfig
.getConfig()
.edit()
- // never drop slot for stream split, which is also global split
- .with(DROP_SLOT_ON_STOP.name(), false)
+ // When startup.mode='snapshot' drop slot for stream split,
+ // which is also global split
Review Comment:
In our current implementation of deleting backfill slots, in snapshot mode, there is no need to restart. You just have to run it again with overwrite mode？ Also Streaming job，It's the same thing.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113#discussion_r2329390546
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java:
##########
@@ -133,6 +135,17 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
}
}
+ @Override
+ protected void onSplitFinished(Map finishedSplitIds) {
+ super.onSplitFinished(finishedSplitIds);
+
+ if (this.sourceConfig.getStartupOptions().isSnapshotOnly()) {
+ PostgresDialect dialect = (PostgresDialect) this.dialect;
+ boolean removed = dialect.removeSlot(dialect.getSlotName());
+ LOG.info("Remove slot '{}' result is {}.", dialect.getSlotName(), removed);
+ }
+ }
+
Review Comment:
```suggestion
@Override
protected void onSplitFinished(Map<String, SourceSplitState> finishedSplitIds) {
super.onSplitFinished(finishedSplitIds);
for (SourceSplitState splitState : finishedSplitIds.values()) {
SourceSplitBase sourceSplit = splitState.toSourceSplit();
if (sourceSplit.isStreamSplit() ) {
StreamSplit streamSplit = sourceSplit.asStreamSplit();
if(streamSplit.getStartingOffset().isAtOrAfter(streamSplit.getEndingOffset())){
PostgresDialect dialect = (PostgresDialect) this.dialect;
boolean removed = dialect.removeSlot(dialect.getSlotName());
LOG.info("Remove slot '{}' result is {}.", dialect.getSlotName(), removed);
}
}
}
}
```
It seems that even a snapshot split is finished, the global slot name will be removed.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java:
##########
@@ -193,16 +191,16 @@ private void maybeCreateSlotForBackFillReadTask(
/** Drop slot for backfill task and close replication connection. */
private void maybeDropSlotForBackFillReadTask(
- PostgresReplicationConnection replicationConnection, boolean skipSnapshotBackfill) {
+ PostgresConnection connection, String slotName, boolean skipSnapshotBackfill) {
// if skip backfill, no need to create slot here
if (skipSnapshotBackfill) {
return;
}
try {
- replicationConnection.close(true);
+ connection.dropReplicationSlot(slotName);
Review Comment:
Why needs change 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113#discussion_r2320705697
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java:
##########
@@ -148,8 +148,11 @@ public void configure(SourceSplitBase sourceSplitBase) {
dbzConfig
.getConfig()
.edit()
- // never drop slot for stream split, which is also global split
- .with(DROP_SLOT_ON_STOP.name(), false)
+ // When startup.mode='snapshot' drop slot for stream split,
+ // which is also global split
Review Comment:
I don't think so, If the job success to finish the stream split, it does no need to restart. However, If fails before complete stream split, it means that the job is not finished, and need to restart with checkpoint .
For batch mode, it has no checkpoint, thus no need to restart with checkpont. But streaming job, it do need. Otherwise, cost is too much.
@LYanquan , WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113#issuecomment-3249145264
@loserwang1024 PTLK
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-38277] [postgres]drop slot when startup.mode=snapshot [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4113:
URL: https://github.com/apache/flink-cdc/pull/4113#discussion_r2358302331
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java:
##########
@@ -133,6 +136,25 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
}
}
+ @Override
+ protected void onSplitFinished(Map finishedSplitIds) {
+ super.onSplitFinished(finishedSplitIds);
+ for (Object splitState : finishedSplitIds.values()) {
+ SourceSplitBase sourceSplit = ((SourceSplitState) splitState).toSourceSplit();
+ if (sourceSplit.isStreamSplit()) {
+ StreamSplit streamSplit = sourceSplit.asStreamSplit();
+ if (this.sourceConfig.getStartupOptions().isSnapshotOnly()
+ || streamSplit
+ .getStartingOffset()
+ .isAtOrAfter(streamSplit.getEndingOffset())) {
Review Comment:
```suggestion
this.sourceConfig.getStartupOptions().isSnapshotOnly()
&& streamSplit
.getStartingOffset()
.isAtOrAfter(streamSplit.getEndingOffset()
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org