lvyanquan commented on code in PR #4550:
URL: https://github.com/apache/flink-cdc/pull/4550#discussion_r4056591769
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java:
##########
@@ -481,6 +501,14 @@ private void requestStreamSplitMetaIfNeeded(StreamSplit
streamSplit) {
} else {
LOG.info("The meta of stream split {} has been collected success",
splitId);
this.addSplits(Collections.singletonList(streamSplit));
+ // Report the stream-split metadata assembled so the coordinator
can release what it no
+ // longer needs to serve. Carries the assignment generation so a
stale report from a
+ // failed attempt is ignored.
+ context.sendSourceEventToCoordinator(
Review Comment:
addSplits(streamSplit) enters the completed stream-split branch above, which
already sends a StreamSplitMetaAssembledEvent. This therefore sends the event
twice on the group-fetch path. Could we remove this duplicate notification and
update the comment in the completed-split branch to also mention metadata
assembled through group fetch? Non-blocking, since the coordinator handling is
idempotent.
##########
docs/content/docs/connectors/flink-sources/db2-cdc.md:
##########
@@ -263,6 +263,13 @@ Db2 server.
If the flink version is greater than or equal to 1.15, the default
value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has
been changed to true,
so it does not need to be explicitly configured
'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
</td>
+ </tr>
+ <tr>
+ <td>scan.incremental.snapshot.metadata.release.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to release the snapshot split metadata (assigned splits,
finished offsets and table schemas) held by the source coordinator once the
source has entered the stream phase, to reduce JobManager memory on jobs with a
very large number of snapshot splits. Disabled by default. Incompatible with
scan.newly-added-table.enabled: enabling both fails at startup, and a job that
has released the metadata cannot later enable newly-added-table scanning.
Release happens only after a successful checkpoint; if checkpointing is
disabled or no checkpoint completes, the metadata is retained, so this option
has no effect without checkpointing.</td>
Review Comment:
Could we document the downgrade restriction for all connectors that expose
this option, including MySQL? Once a checkpoint or savepoint is created with
this option enabled, it cannot be used to restore the job after downgrading to
Flink CDC 3.6.0 or an earlier version. MySQL has the same state compatibility
restriction and exposes this option in both the Flink Source and Pipeline
Connector APIs, so the corresponding English and Chinese MySQL documentation
should be updated as well. This is non-blocking.
--
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]
