Fix bug in Java9IntHash calculation that assumed crc32c_update(x) == …

…~crc32c_update(~x)

- Java9IntHash uses private methods from java.util.zip.CRC32C class,
  updateBytes and updateDirectByteBuffer.
  When inspecting the use and interface contract, it doesn't match
  how it is used in Java9IntHash. This PR addresses that by introducing
  a separate initial value for initializing the accumulated value
  so that the initial value could match the logic in
  java.util.zip.CRC32C.reset method. There's also a separate
  method for finalizing the accumulated value into a final
  checksum value. This is to match the java.util.zip.CRC32C.getValue
  method's logic (uses bitwise complement operator ~).

- With a quick glance, it might appear that the previous logic is similar.
  However it isn't since I have a failing test which gets fixed with this
  change. I haven't yet added the Java9IntHash level unit test case to prove how
  it differs. It must be related to integer value overflow. For the CRC32C function,
  I believe it means that it cannot be assumed in all cases that
  func(x) == ~func(~x). That's the assumption that the previous code was making.
  It probably applies for many inputs, but not all. It would break in overflow
  cases.

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32CDigestManager.java

@@ -54,4 +54,14 @@ int internalUpdate(int digest, ByteBuf data, int offset, int len) {
     int internalUpdate(int digest, byte[] buffer, int offset, int len) {
         return Crc32cIntChecksum.resumeChecksum(digest, buffer, offset, len);
     }
-}
+
+    @Override
+    protected int initialDigest() {
+        return Crc32cIntChecksum.initialValue();
+    }
+
+    @Override
+    protected int finalizeDigest(int digest) {
+        return Crc32cIntChecksum.finalizeValue(digest);
+    }
+}

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java

@@ -61,12 +61,31 @@ public abstract class DigestManager {
 
     abstract int internalUpdate(int digest, byte[] buffer, int offset, int len);
 
+    protected int initialDigest() {
+        return 0;
+    };
+
+    protected int finalizeDigest(int digest) {
+        return digest;
+    }
+
     final int update(int digest, ByteBuf buffer, int offset, int len) {
+        if (len == 0) {
+            return digest;
+        }
         MutableInt digestRef = new MutableInt(digest);
         ByteBufVisitor.visitBuffers(buffer, offset, len, byteBufVisitorCallback, digestRef);
         return digestRef.intValue();
     }
 
+    final protected int initializeAndUpdate(ByteBuf buf, int offset, int len) {
+        return update(initialDigest(), buf, offset, len);
+    }
+
+    private int updateAndFinalize(ByteBuf data, int digest, int offset, int len) {
+        return finalizeDigest(update(digest, data, offset, len));
+    }
+
     abstract void populateValueAndReset(int digest, ByteBuf buffer);
 
     abstract boolean isInt32Digest();
@@ -145,8 +164,8 @@ private ReferenceCounted computeDigestAndPackageForSendingV2(long entryId, long
         buf.writeLong(length);
 
         // Compute checksum over the headers
-        int digest = update(0, buf, buf.readerIndex(), buf.readableBytes());
-        digest = update(digest, data, data.readerIndex(), data.readableBytes());
+        int digest = initializeAndUpdate(buf, buf.readerIndex(), buf.readableBytes());
+        digest = updateAndFinalize(data, digest, data.readerIndex(), data.readableBytes());
 
         populateValueAndReset(digest, buf);
 
@@ -170,8 +189,8 @@ private ByteBufList computeDigestAndPackageForSendingV3(long entryId, long lastA
         headersBuffer.writeLong(lastAddConfirmed);
         headersBuffer.writeLong(length);
 
-        int digest = update(0, headersBuffer, 0, METADATA_LENGTH);
-        digest = update(digest, data, data.readerIndex(), data.readableBytes());
+        int digest = initializeAndUpdate(headersBuffer, 0, METADATA_LENGTH);
+        digest = updateAndFinalize(data, digest, data.readerIndex(), data.readableBytes());
         populateValueAndReset(digest, headersBuffer);
         return ByteBufList.get(headersBuffer, data);
     }
@@ -193,7 +212,7 @@ public ByteBufList computeDigestAndPackageForSendingLac(long lac) {
         headersBuffer.writeLong(ledgerId);
         headersBuffer.writeLong(lac);
 
-        int digest = update(0, headersBuffer, 0, LAC_METADATA_LENGTH);
+        int digest = finalizeDigest(initializeAndUpdate(headersBuffer, 0, LAC_METADATA_LENGTH));
         populateValueAndReset(digest, headersBuffer);
 
         return ByteBufList.get(headersBuffer);
@@ -229,10 +248,10 @@ private void verifyDigest(long entryId, ByteBuf dataReceived, boolean skipEntryI
                     this.getClass().getName(), dataReceived.readableBytes());
             throw new BKDigestMatchException();
         }
-        int digest = update(0, dataReceived, 0, METADATA_LENGTH);
+        int digest = initializeAndUpdate(dataReceived, 0, METADATA_LENGTH);
 
         int offset = METADATA_LENGTH + macCodeLength;
-        digest = update(digest, dataReceived, offset, dataReceived.readableBytes() - offset);
+        digest = updateAndFinalize(dataReceived, digest, offset, dataReceived.readableBytes() - offset);
 
         if (isInt32Digest()) {
             int receivedDigest = dataReceived.getInt(METADATA_LENGTH);
@@ -277,7 +296,7 @@ public long verifyDigestAndReturnLac(ByteBuf dataReceived) throws BKDigestMatchE
             throw new BKDigestMatchException();
         }
 
-        int digest = update(0, dataReceived, 0, LAC_METADATA_LENGTH);
+        int digest = finalizeDigest(initializeAndUpdate(dataReceived, 0, LAC_METADATA_LENGTH));
 
         if (isInt32Digest()) {
             int receivedDigest = dataReceived.getInt(LAC_METADATA_LENGTH);

bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/checksum/CompositeByteBufUnwrapBugReproduceTest.java

@@ -129,6 +129,16 @@ int internalUpdate(int digest, ByteBuf data, int offset, int len) {
         int internalUpdate(int digest, byte[] buffer, int offset, int len) {
             return intHash.resume(digest, buffer, offset, len);
         }
+
+        @Override
+        protected int initialDigest() {
+            return intHash.initialValue();
+        }
+
+        @Override
+        protected int finalizeDigest(int digest) {
+            return intHash.finalizeValue(digest);
+        }
     }
 
     @Test

circe-checksum/src/main/java/com/scurrilous/circe/checksum/Crc32cIntChecksum.java

@@ -94,4 +94,11 @@ public static int resumeChecksum(int previousChecksum, byte[] payload, int offse
         return CRC32C_HASH.resume(previousChecksum, payload, offset, len);
     }
 
+    public static int initialValue() {
+        return CRC32C_HASH.initialValue();
+    }
+
+    public static int finalizeValue(int current) {
+        return CRC32C_HASH.finalizeValue(current);
+    }
 }

circe-checksum/src/main/java/com/scurrilous/circe/checksum/IntHash.java

@@ -30,4 +30,12 @@ public interface IntHash {
     int resume(int current, ByteBuf buffer, int offset, int len);
 
     int resume(int current, byte[] buffer, int offset, int len);
+
+    default int initialValue() {
+        return 0;
+    }
+
+    default int finalizeValue(int current) {
+        return current;
+    }
 }

circe-checksum/src/main/java/com/scurrilous/circe/checksum/Java9IntHash.java

@@ -68,12 +68,12 @@ protected byte[] initialValue() {
 
     @Override
     public int calculate(ByteBuf buffer) {
-        return resume(0, buffer);
+        return finalizeValue(resume(initialValue(), buffer));
     }
 
     @Override
     public int calculate(ByteBuf buffer, int offset, int len) {
-        return resume(0, buffer, offset, len);
+        return finalizeValue(resume(initialValue(), buffer, offset, len));
     }
 
     private int resume(int current, long address, int offset, int length) {
@@ -100,26 +100,35 @@ public int resume(int current, ByteBuf buffer) {
 
     @Override
     public int resume(int current, ByteBuf buffer, int offset, int len) {
-        int negCrc = ~current;
-
         if (buffer.hasMemoryAddress()) {
-            negCrc = resume(negCrc, buffer.memoryAddress(), offset, len);
+            current = resume(current, buffer.memoryAddress(), offset, len);
         } else if (buffer.hasArray()) {
             int arrayOffset = buffer.arrayOffset() + offset;
-            negCrc = resume(negCrc, buffer.array(), arrayOffset, len);
+            current = resume(current, buffer.array(), arrayOffset, len);
         } else {
             byte[] b = TL_BUFFER.get();
             int toRead = len;
             int loopOffset = offset;
             while (toRead > 0) {
                 int length = Math.min(toRead, b.length);
                 buffer.slice(loopOffset, length).readBytes(b, 0, length);
-                negCrc = resume(negCrc, b, 0, length);
+                current = resume(current, b, 0, length);
                 toRead -= length;
                 loopOffset += length;
             }
         }
+        return current;
+    }
+
+    @Override
+    public int initialValue() {
+        // match java.util.zip.CRC32C.reset logic
+        return 0xFFFFFFFF;
+    }
 
-        return ~negCrc;
+    @Override
+    public int finalizeValue(int current) {
+        // match java.util.zip.CRC32C.getValue logic
+        return ~current;
     }
 }

circe-checksum/src/test/java/com/scurrilous/circe/checksum/ChecksumTest.java

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package com.scurrilous.circe.checksum;
 
 import static com.scurrilous.circe.params.CrcParameters.CRC32C;
@@ -44,7 +45,9 @@ public void testCrc32cValue() {
     @Test
     public void testCrc32cValueResume() {
         final byte[] bytes = "Some String".getBytes();
-        int checksum = Crc32cIntChecksum.resumeChecksum(0, Unpooled.wrappedBuffer(bytes), 0, bytes.length);
+        int checksum = Crc32cIntChecksum.finalizeValue(
+                Crc32cIntChecksum.resumeChecksum(Crc32cIntChecksum.initialValue(), Unpooled.wrappedBuffer(bytes), 0,
+                        bytes.length));
 
         assertEquals(608512271, checksum);
     }
@@ -56,22 +59,26 @@ public void testCrc32cValueIncremental() {
         int checksum = Crc32cIntChecksum.computeChecksum(Unpooled.wrappedBuffer(bytes));
         assertEquals(608512271, checksum);
 
-        checksum = Crc32cIntChecksum.computeChecksum(Unpooled.wrappedBuffer(bytes, 0, 1));
+        checksum =
+                Crc32cIntChecksum.resumeChecksum(Crc32cIntChecksum.initialValue(), Unpooled.wrappedBuffer(bytes, 0, 1));
         for (int i = 1; i < bytes.length; i++) {
             checksum = Crc32cIntChecksum.resumeChecksum(checksum, Unpooled.wrappedBuffer(bytes), i, 1);
         }
+        checksum = Crc32cIntChecksum.finalizeValue(checksum);
         assertEquals(608512271, checksum);
 
-        checksum = Crc32cIntChecksum.computeChecksum(Unpooled.wrappedBuffer(bytes, 0, 4));
+        checksum =
+                Crc32cIntChecksum.resumeChecksum(Crc32cIntChecksum.initialValue(), Unpooled.wrappedBuffer(bytes, 0, 4));
         checksum = Crc32cIntChecksum.resumeChecksum(checksum, Unpooled.wrappedBuffer(bytes), 4, 7);
+        checksum = Crc32cIntChecksum.finalizeValue(checksum);
         assertEquals(608512271, checksum);
 
 
         ByteBuf buffer = Unpooled.wrappedBuffer(bytes, 0, 4);
-        checksum = Crc32cIntChecksum.computeChecksum(buffer);
+        checksum = Crc32cIntChecksum.resumeChecksum(Crc32cIntChecksum.initialValue(), buffer);
         checksum = Crc32cIntChecksum.resumeChecksum(
-            checksum, Unpooled.wrappedBuffer(bytes), 4, bytes.length - 4);
-
+                checksum, Unpooled.wrappedBuffer(bytes), 4, bytes.length - 4);
+        checksum = Crc32cIntChecksum.finalizeValue(checksum);
         assertEquals(608512271, checksum);
     }
 
@@ -86,7 +93,10 @@ public void testCrc32cLongValue() {
     @Test
     public void testCrc32cLongValueResume() {
         final byte[] bytes = "Some String".getBytes();
-        long checksum = Crc32cIntChecksum.resumeChecksum(0, Unpooled.wrappedBuffer(bytes), 0, bytes.length);
+        long checksum =
+                Crc32cIntChecksum.resumeChecksum(Crc32cIntChecksum.initialValue(), Unpooled.wrappedBuffer(bytes), 0,
+                        bytes.length);
+        checksum = Crc32cIntChecksum.finalizeValue((int) checksum);
 
         assertEquals(608512271L, checksum);
     }

circe-checksum/src/test/java/com/scurrilous/circe/checksum/Java9IntHashTest.java

@@ -65,8 +65,8 @@ public void calculateCheckSumUsingCompositeByteBuf() {
         int checksumRes1 = Crc32cIntChecksum.computeChecksum(bTotal);
 
         // Calculate: case-2.
-        int b1CheckSum = Crc32cIntChecksum.computeChecksum(b1);
-        int checksumRes2 = Crc32cIntChecksum.resumeChecksum(b1CheckSum, b2);
+        int b1CheckSum = Crc32cIntChecksum.resumeChecksum(Crc32cIntChecksum.initialValue(), b1);
+        int checksumRes2 = Crc32cIntChecksum.finalizeValue(Crc32cIntChecksum.resumeChecksum(b1CheckSum, b2));
 
         // Verify: the results of both ways to calculate the checksum are same.
         Assert.assertEquals(checksumRes1, checksumRes2);
@@ -90,8 +90,8 @@ public void calculateCheckSumUsingNoArrayNoMemoryAddrByteBuf() {
         int checksumRes1 = Crc32cIntChecksum.computeChecksum(bTotal);
 
         // Calculate: case-2.
-        int b1CheckSum = Crc32cIntChecksum.computeChecksum(b1);
-        int checksumRes2 = Crc32cIntChecksum.resumeChecksum(b1CheckSum, b2);
+        int b1CheckSum = Crc32cIntChecksum.resumeChecksum(Crc32cIntChecksum.initialValue(), b1);
+        int checksumRes2 = Crc32cIntChecksum.finalizeValue(Crc32cIntChecksum.resumeChecksum(b1CheckSum, b2));
 
         // Verify: the results of both ways to calculate the checksum are same.
         Assert.assertEquals(checksumRes1, checksumRes2);