[GitHub] [flink] xintongsong commented on a diff in pull request #20647: [FLINK-29053] Hybrid shuffle has concurrent modification of buffer when compression is enabled

[GitBox]

xintongsong commented on code in PR #20647:
URL: https://github.com/apache/flink/pull/20647#discussion_r956152093


##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java:
##########
@@ -141,7 +141,9 @@ public ReadOnlySlicedNetworkBuffer readOnlySlice() {
 
     @Override
     public ReadOnlySlicedNetworkBuffer readOnlySlice(int index, int length) {
-        checkState(!isCompressed, "Unable to slice a compressed buffer.");
+        checkState(
+                !isCompressed || index + length == writerIndex(),
+                "Unable to slice a partial compressed buffer.");
         return new ReadOnlySlicedNetworkBuffer(
                 super.unwrap(), index, length, memorySegmentOffset, false);

Review Comment:
   ```suggestion
                   super.unwrap(), index, length, memorySegmentOffset, 
isCompressed);
   ```
   
   We should do the same for `NetworkBuffer#readOnlySlice`, by adding a new 
argument to the constructor of`ReadOnlySlicedNetworkBuffer`.
   
   Then you won't need to call `setCompressed` in 
`HsSubpartitionMemoryDataManager#getSliceBuffer`.



##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java:
##########
@@ -173,7 +173,9 @@ public ReadOnlySlicedNetworkBuffer readOnlySlice() {
 
     @Override
     public ReadOnlySlicedNetworkBuffer readOnlySlice(int index, int length) {
-        checkState(!isCompressed, "Unable to slice a compressed buffer.");
+        checkState(
+                !isCompressed || index + length != writerIndex(),
+                "Unable to slice a partial compressed buffer.");

Review Comment:
   ```suggestion
                   "Unable to partially slice a compressed buffer.");
   ```



##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionMemoryDataManager.java:
##########
@@ -359,13 +364,29 @@ private void finishCurrentWritingBuffer() {
         Buffer buffer = bufferConsumer.build();
         currentWritingBuffer.close();
         bufferConsumer.close();
-
+        // TODO support buffer compression

Review Comment:
   What is this TODO for?



##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java:
##########
@@ -141,7 +141,9 @@ public ReadOnlySlicedNetworkBuffer readOnlySlice() {
 
     @Override
     public ReadOnlySlicedNetworkBuffer readOnlySlice(int index, int length) {
-        checkState(!isCompressed, "Unable to slice a compressed buffer.");
+        checkState(
+                !isCompressed || index + length == writerIndex(),
+                "Unable to slice a partial compressed buffer.");

Review Comment:
   ```suggestion
                   "Unable to partially slice a compressed buffer.");
   ```



-- 
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-unsubscr...@flink.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org