lvyanquan commented on code in PR #3966:
URL: https://github.com/apache/flink-cdc/pull/3966#discussion_r2011620652
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriter.java:
##########
@@ -92,21 +96,29 @@ public PaimonWriter(
}
@Override
- public Collection<MultiTableCommittable> prepareCommit() throws
IOException {
+ public Collection<MultiTableCommittable> prepareCommit() {
List<MultiTableCommittable> committables = new ArrayList<>();
- for (Map.Entry<Identifier, StoreSinkWrite> entry : writes.entrySet()) {
- Identifier key = entry.getKey();
- StoreSinkWrite write = entry.getValue();
- boolean waitCompaction = true;
- committables.addAll(
- // here we set it to lastCheckpointId+1 to
- // avoid prepareCommit the same checkpointId with the
first round.
- write.prepareCommit(waitCompaction, lastCheckpointId +
1).stream()
- .map(
- committable ->
-
MultiTableCommittable.fromCommittable(key, committable))
- .collect(Collectors.toList()));
- }
+ writes.entrySet().parallelStream()
+ .forEach(
Review Comment:
Done.
--
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]
