[apache#804] improvement: Optimize CRC calculation of ByteBuffer

common/src/main/java/org/apache/uniffle/common/util/ChecksumUtils.java

@@ -40,14 +40,19 @@ public static long getCrc32(byte[] buf, int offset, int length) {
     return crc32.getValue();
   }
 
-  // you may need to flip at first
   public static long getCrc32(ByteBuffer byteBuffer) {
-    if (byteBuffer.hasArray()) {
-      return getCrc32(byteBuffer.array(), byteBuffer.arrayOffset() + byteBuffer.position(), byteBuffer.remaining());
-    } else {
-      byte[] byteArray = new byte[byteBuffer.remaining()];
-      byteBuffer.get(byteArray);
-      return getCrc32(byteArray);
+    return getCrc32(byteBuffer, byteBuffer.position(), byteBuffer.limit() - byteBuffer.position());
+  }
+
+  public static long getCrc32(ByteBuffer byteBuffer, int offset, int length) {
+    CRC32 crc32 = new CRC32();
+    byteBuffer.position(offset);
+    for (int i = 0; i < length; ) {
+      int len = Math.min(LENGTH_PER_CRC, length -i);
+      byteBuffer.limit(byteBuffer.position() + len);
+      crc32.update(byteBuffer);
+      i += len;
     }
+    return crc32.getValue();
   }
 }

common/src/test/java/org/apache/uniffle/common/util/ChecksumUtilsTest.java

@@ -89,4 +89,25 @@ public void crc32TestWithByteBuff() throws Exception {
     assertEquals(expectedChecksum, ChecksumUtils.getCrc32(buffer));
 
   }
+
+  @Test
+  public void crc32ByteBufferTest() throws Exception {
+    int length = 32 * 1024 * 1024;
+    byte[] data = new byte[length];
+    Random random = new Random();
+    random.nextBytes(data);
+    long expectCrc = ChecksumUtils.getCrc32(data);
+    assertEquals(expectCrc, ChecksumUtils.getCrc32(ByteBuffer.wrap(data)));
+    ByteBuffer directBuffer = ByteBuffer.allocateDirect(length);
+    directBuffer.put(data);
+    directBuffer.flip();
+    assertEquals(expectCrc, ChecksumUtils.getCrc32(directBuffer));
+    int offset = random.nextInt(15);
+    ByteBuffer directOffsetBuffer = ByteBuffer.allocateDirect(length + offset);
+    byte[] dataOffset = new byte[offset];
+    random.nextBytes(dataOffset);
+    directOffsetBuffer.put(dataOffset);
+    directOffsetBuffer.put(data);
+    assertEquals(expectCrc, ChecksumUtils.getCrc32(directOffsetBuffer, offset, length));
+  }
 }