blobstore: checksum support for mpu

Author: sandeepvinayak

blob/blob-inmemory/src/main/java/com/salesforce/multicloudj/blob/inmemory/InMemoryBlobStore.java

@@ -591,6 +591,7 @@ protected MultipartUpload doInitiateMultipartUpload(MultipartUploadRequest reque
         .key(request.getKey())
         .metadata(request.getMetadata())
         .tags(request.getTags())
+        .checksumEnabled(request.isChecksumEnabled())
         .kmsKeyId(request.getKmsKeyId())
         .build();
   }
@@ -613,10 +614,11 @@ protected UploadPartResponse doUploadMultipartPart(MultipartUpload mpu, Multipar
 
       byte[] partData = baos.toByteArray();
       String etag = generateEtag(partData);
+      String checksumValue = computeCrc32cChecksum(partData);
 
       state.addPart(mpp.getPartNumber(), partData, etag);
 
-      return new UploadPartResponse(mpp.getPartNumber(), etag, (long) partData.length);
+      return new UploadPartResponse(mpp.getPartNumber(), etag, partData.length, checksumValue);
     } catch (Exception e) {
       throw new UnknownException("Failed to upload multipart part", e);
     }
@@ -675,8 +677,9 @@ protected MultipartUploadResponse doCompleteMultipartUpload(
       STORAGE.put(versionedKey, blob);
       LATEST_VERSIONS.put(baseKey, versionId);
       MULTIPART_UPLOADS.remove(mpu.getId());
+      String checksumValue = computeCrc32cChecksum(finalData);
 
-      return new MultipartUploadResponse(etag);
+      return new MultipartUploadResponse(etag, checksumValue);
     } catch (Exception e) {
       throw new UnknownException("Failed to complete multipart upload", e);
     }
@@ -696,7 +699,7 @@ protected List<UploadPartResponse> doListMultipartUpload(MultipartUpload mpu) {
                 new UploadPartResponse(
                     entry.getKey(),
                     entry.getValue().getEtag(),
-                    (long) entry.getValue().getData().length))
+                    entry.getValue().getData().length))
         .sorted(Comparator.comparingInt(UploadPartResponse::getPartNumber))
         .collect(Collectors.toList());
   }
@@ -788,6 +791,18 @@ private String generateEtag(byte[] data) {
     return "\"" + Integer.toHexString(java.util.Arrays.hashCode(data)) + "\"";
   }
 
+  private String computeCrc32cChecksum(byte[] data) {
+    java.util.zip.CRC32C crc32c = new java.util.zip.CRC32C();
+    crc32c.update(data);
+    long value = crc32c.getValue();
+    byte[] checksumBytes = new byte[4];
+    checksumBytes[0] = (byte) (value >> 24);
+    checksumBytes[1] = (byte) (value >> 16);
+    checksumBytes[2] = (byte) (value >> 8);
+    checksumBytes[3] = (byte) value;
+    return java.util.Base64.getEncoder().encodeToString(checksumBytes);
+  }
+
   private byte[] extractRange(byte[] data, Long start, Long end) {
     int dataLength = data.length;