[GitHub] [flink] akalash commented on a diff in pull request #22761: [FLINK-32298][network] Fix the bug that totalWrittenBytes of BufferWritingResultPartition misses some data

["akalash (via GitHub)" <gi...@apache.org>]

akalash commented on code in PR #22761:
URL: https://github.com/apache/flink/pull/22761#discussion_r1262673283


##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java:
##########
@@ -65,7 +65,7 @@ public abstract class BufferWritingResultPartition extends ResultPartition {
 
     private TimerGauge hardBackPressuredTimeMsPerSecond = new TimerGauge();
 
-    private long totalWrittenBytes;
+    protected long totalWrittenBytes;

Review Comment:
   Ideally, `totalWrittenBytes` should be changed only in one class `BufferWritingResultPartition`. So maybe we should forbid `inherited classes` to write to `subpartitions` directly. But honestly, I don't know how to do it easily since the current implementation of `inherited classes` use `subpartitions` a lot and we can not change `subpartitions` visibility from `protected` to `private` . I mean we can create `addToSubpartition` method in `BufferWritingResultPartition` and use it everywhere(and maybe it will be better than now) but it still doesn't protect us from possible future bugs if somebody decides to use `subpartitions#add` directly in `inherited classes`.
   
   Anyway, if you don't think that it will make things worse, maybe we indeed for now create the `add`(or `addToSubpartition`) method in `BufferWritingResultPartition` and will use it everywhere. At least it will reduce the number of places where we update `totalWrittenBytes`. Unfortunatelly, I didn't come up with any other solution until now.



##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java:
##########
@@ -190,51 +195,60 @@ public void recover(
             int oldSubtaskIndex,
             BufferWithContext<BufferBuilder> bufferWithContext)
             throws IOException {
-        try (BufferBuilder bufferBuilder = bufferWithContext.context) {
-            try (BufferConsumer bufferConsumer =
-                    bufferBuilder.createBufferConsumerFromBeginning()) {
-                bufferBuilder.finish();
-                if (bufferConsumer.isDataAvailable()) {
-                    final List<CheckpointedResultSubpartition> channels =
-                            getMappedChannels(subpartitionInfo);
-                    for (final CheckpointedResultSubpartition channel : channels) {
-                        // channel selector is created from the downstream's point of view: the
-                        // subtask of downstream = subpartition index of recovered buffer
-                        final SubtaskConnectionDescriptor channelSelector =
-                                new SubtaskConnectionDescriptor(
-                                        subpartitionInfo.getSubPartitionIdx(), oldSubtaskIndex);
-                        channel.addRecovered(
-                                EventSerializer.toBufferConsumer(channelSelector, false));
-                        channel.addRecovered(bufferConsumer.copy());
-                    }
-                }
+        try (BufferBuilder bufferBuilder = bufferWithContext.context;
+                BufferConsumer bufferConsumer = bufferBuilder.createBufferConsumerFromBeginning()) {
+            bufferBuilder.finish();
+            if (!bufferConsumer.isDataAvailable()) {
+                return;
+            }
+            final List<ResultSubpartitionInfo> mappedSubpartitions =
+                    getMappedSubpartitions(subpartitionInfo);
+            for (final ResultSubpartitionInfo mappedSubpartition : mappedSubpartitions) {
+                // channel selector is created from the downstream's point of view: the
+                // subtask of downstream = subpartition index of recovered buffer
+                final SubtaskConnectionDescriptor channelSelector =
+                        new SubtaskConnectionDescriptor(
+                                subpartitionInfo.getSubPartitionIdx(), oldSubtaskIndex);
+                CheckpointedResultPartition checkpointedResultPartition =
+                        getCheckpointedResultPartition(mappedSubpartition.getPartitionIdx());
+                checkpointedResultPartition.addRecovered(

Review Comment:
   good. It's better now



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