[GitHub] [flink] curcur commented on a diff in pull request #21405: [FLINK-23035][state/changelog] Add explicit append() to StateChangelogWriter to write metadata

[GitBox <gi...@apache.org>]

curcur commented on code in PR #21405:
URL: https://github.com/apache/flink/pull/21405#discussion_r1041764339


##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/StateChange.java:
##########
@@ -26,14 +26,21 @@
 @Internal
 public class StateChange implements Serializable {
 
+    /* For metadata, see FLINK-23035.*/
+    public static final int COMMON_KEY_GROUP = -1;

Review Comment:
   rename this to META_KEY_GROUP?



##########
flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/FsStateChangelogWriter.java:
##########
@@ -167,6 +167,20 @@ class FsStateChangelogWriter implements StateChangelogWriter<ChangelogStateHandl
         this.localChangelogRegistry = localChangelogRegistry;
     }
 
+    @Override
+    public void append(byte[] value) throws IOException {
+        LOG.trace("append metadata to {}: {} bytes", logId, value.length);
+        checkState(!closed, "%s is closed", logId);
+        activeChangeSet.add(new StateChange(value));
+        activeChangeSetSize += value.length;
+        if (activeChangeSetSize >= preEmptivePersistThresholdInBytes) {
+            LOG.debug(
+                    "pre-emptively flush {}MB of appended changes to the common store",
+                    activeChangeSetSize / 1024 / 1024);
+            persistInternal(notUploaded.isEmpty() ? activeSequenceNumber : notUploaded.firstKey());
+        }
+    }

Review Comment:
   There are a lot of duplicate code between the two "append" methods.
   
   Please extract the two common part and rename this to "appendMeta" for example.



-- 
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