fredia commented on code in PR #21822:
URL: https://github.com/apache/flink/pull/21822#discussion_r1140928577
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogStorage.java:
##########
@@ -167,6 +167,7 @@ public FsStateChangelogWriter createWriter(
@Override
public void close() throws Exception {
uploader.close();
+ localChangelogRegistry.close();
Review Comment:
Fixed,replace with `IOUtils.closeQuietly` .
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java:
##########
@@ -199,15 +200,21 @@ public SequenceNumber nextSequenceNumber() {
return activeSequenceNumber;
}
+ @VisibleForTesting
+ CompletableFuture<SnapshotResult<ChangelogStateHandleStreamImpl>>
persist(SequenceNumber from)
Review Comment:
The tests related to the Local file have been modified accordingly, for
other tests I left them as they are.
##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java:
##########
@@ -369,7 +390,6 @@ public void confirm(SequenceNumber from, SequenceNumber to,
long checkpointId) {
changelogRegistry.stopTracking(localHandle);
localChangelogRegistry.register(localHandle,
checkpointId);
Review Comment:
Fixed.
--
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]
