fix(api): change digest value to string

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/FileObjectMeta.java

@@ -81,7 +81,7 @@ default ConnectHeaders toConnectHeader() {
         ConnectHeaders headers = new ConnectHeaders();
         headers.addString("connect.file.name", name());
         headers.addString("connect.file.uri", uri().toString());
-        headers.addLong("connect.file.hash.digest", contentDigest().digest());
+        headers.addString("connect.file.hash.digest", contentDigest().digest());
         headers.addString("connect.file.hash.algorithm", contentDigest().algorithm());
         headers.addLong("connect.file.contentLength", contentLength());
         headers.addLong("connect.file.lastModified", lastModified());
@@ -96,18 +96,18 @@ default int compareTo(final FileObjectMeta that) {
     }
 
     class ContentDigest {
-        private final long digest;
+        private final String digest;
         private final String algorithm;
 
         @JsonCreator
-        public ContentDigest(@JsonProperty("digest") final long digest,
+        public ContentDigest(@JsonProperty("digest") final String digest,
                              @JsonProperty("algorithm") final String algorithm) {
-            this.digest = digest;
+            this.digest = Objects.requireNonNull(digest, "digest should not be null");;
             this.algorithm = Objects.requireNonNull(algorithm, "algorithm should not be null");
         }
 
         @JsonProperty("digest")
-        public long digest() {
+        public String digest() {
             return digest;
         }
 
@@ -121,7 +121,7 @@ public boolean equals(Object o) {
             if (this == o) return true;
             if (!(o instanceof ContentDigest)) return false;
             ContentDigest that = (ContentDigest) o;
-            return digest == that.digest &&
+            return Objects.equals(digest, that.digest) &&
                     Objects.equals(algorithm, that.algorithm);
         }
 

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/LocalFileObjectMeta.java

@@ -117,9 +117,9 @@ private static ContentDigest hash(final File f) {
                 byte[] bytes = readStartingBytesFrom(f, 4096);
                 crc32.update(bytes);
                 crc32.update(longToBytes(f.length()));
-                return new ContentDigest(crc32.getValue(), "crc32");
+                return new ContentDigest(String.valueOf(crc32.getValue()), "crc32");
             }
-            return new ContentDigest(-1, "crc32");
+            return new ContentDigest("", "crc32");
         } catch (IOException e) {
             throw new ConnectFilePulseException(
                     "Error while computing CRC32 hash for file : " + f.getName() + " - " + e.getLocalizedMessage());

connect-file-pulse-plugin/src/test/java/io/streamthoughts/kafka/connect/filepulse/offset/DefaultOffsetPolicyTest.java

@@ -37,7 +37,7 @@ public class DefaultOffsetPolicyTest {
             "test",
             0L,
             123L,
-            new FileObjectMeta.ContentDigest(789L, "dummy"),
+            new FileObjectMeta.ContentDigest("789", "dummy"),
             Collections.singletonMap(LocalFileObjectMeta.SYSTEM_FILE_INODE_META_KEY, "456L")
     );
 
@@ -67,7 +67,7 @@ public void should_get_offset_based_on_path() {
     public void should_get_offset_based_on_hash() {
         Map<String, Object> result = new DefaultOffsetPolicy("HASH").toPartitionMap(metadata);
         Assert.assertEquals(1, result.size());
-        Assert.assertEquals(789L, result.get("hash"));
+        Assert.assertEquals("789", result.get("hash"));
     }
 
     @Test
@@ -91,7 +91,7 @@ public void should_get_composed_offset_based_on_path_and_hash() {
         Map<String, Object> result = new DefaultOffsetPolicy("PATH+HASH").toPartitionMap(metadata);
         Assert.assertEquals(2, result.size());
         Assert.assertEquals("/tmp/path", result.get("path"));
-        Assert.assertEquals(789L, result.get("hash"));
+        Assert.assertEquals("789", result.get("hash"));
     }
 
     @Test

connect-file-pulse-plugin/src/test/java/io/streamthoughts/kafka/connect/filepulse/state/FileObjectSerdeTest.java

@@ -37,7 +37,6 @@ public class FileObjectSerdeTest {
     @Rule
     public TemporaryFolder testFolder = new TemporaryFolder();
 
-
     @Test
     public void should_serialize_and_deserialize_given_valid_object() throws IOException {
         FileObjectSerde serde = new FileObjectSerde();