Use SHA for BLOB update instead of modification time

storm-client/src/jvm/org/apache/storm/blobstore/BlobStore.java

@@ -444,7 +444,7 @@ public BlobStoreFileInputStream(BlobStoreFile part) throws IOException {
 
         @Override
         public long getVersion() throws IOException {
-            return part.getModTime();
+            return part.getVersion();
         }
 
         @Override

storm-client/src/jvm/org/apache/storm/blobstore/BlobStoreFile.java

@@ -42,6 +42,10 @@ public abstract class BlobStoreFile {
 
     public abstract long getModTime() throws IOException;
 
+    public long getVersion() throws IOException {
+        return getModTime();
+    }
+
     public abstract InputStream getInputStream() throws IOException;
 
     public abstract OutputStream getOutputStream() throws IOException;

storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java

@@ -291,7 +291,7 @@ public ReadableBlobMeta getBlobMeta(String key, Subject who) throws Authorizatio
         rbm.set_settable(meta);
         try {
             LocalFsBlobStoreFile pf = fbs.read(DATA_PREFIX + key);
-            rbm.set_version(pf.getModTime());
+            rbm.set_version(pf.getVersion());
         } catch (IOException e) {
             throw new RuntimeException(e);
         }

storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStoreFile.java

@@ -20,8 +20,10 @@
 import java.io.OutputStream;
 import java.nio.file.Files;
 import java.nio.file.StandardCopyOption;
+import java.util.Arrays;
 import java.util.regex.Matcher;
 import org.apache.storm.generated.SettableBlobMeta;
+import org.apache.storm.shade.org.apache.commons.codec.digest.DigestUtils;
 
 public class LocalFsBlobStoreFile extends BlobStoreFile {
 
@@ -72,6 +74,14 @@ public String getKey() {
         return key;
     }
 
+    @Override
+    public long getVersion() throws IOException {
+        try (FileInputStream fis = new FileInputStream(path)) {
+            byte[] bytes = DigestUtils.sha1(fis);
+            return Arrays.hashCode(bytes);
+        }
+    }
+
     @Override
     public long getModTime() throws IOException {
         return path.lastModified();

storm-server/src/test/java/org/apache/storm/blobstore/LocalFsBlobStoreFileTest.java

@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.  The ASF licenses this file to you under the Apache License, Version
+ * 2.0 (the "License"); you may not use this file except in compliance with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
+ * and limitations under the License.
+ */
+
+package org.apache.storm.blobstore;
+
+import org.apache.storm.shade.org.apache.commons.codec.digest.DigestUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class LocalFsBlobStoreFileTest {
+
+    private File tempFile;
+    private LocalFsBlobStoreFile blobStoreFile;
+
+    @BeforeEach
+    public void setUp() throws IOException {
+        tempFile = Files.createTempFile(null, ".tmp").toFile();
+        try (FileOutputStream fs = new FileOutputStream(tempFile)) {
+            fs.write("Content for SHA hash".getBytes());
+        }
+        blobStoreFile = new LocalFsBlobStoreFile(tempFile.getParentFile(), tempFile.getName());
+    }
+
+    @Test
+    void testGetVersion() throws IOException {
+        long expectedVersion = Arrays.hashCode(DigestUtils.sha1("Content for SHA hash"));
+        long actualVersion = blobStoreFile.getVersion();
+        assertEquals(expectedVersion, actualVersion, "The version should match the expected hash code.");
+    }
+
+    @Test
+    void testGetVersion_Mismatch() throws IOException {
+        long expectedVersion = Arrays.hashCode(DigestUtils.sha1("Different content"));
+        long actualVersion = blobStoreFile.getVersion();
+        assertNotEquals(expectedVersion, actualVersion, "The version shouldn't match the hash code of different content.");
+    }
+
+    @Test
+    void testGetVersion_FileNotFound() {
+        boolean deleted = tempFile.delete();
+        if (!deleted) {
+            throw new IllegalStateException("Failed to delete the temporary file.");
+        }
+        assertThrows(IOException.class, () -> blobStoreFile.getVersion(), "Should throw IOException if file is not found.");
+    }
+
+    @Test
+    void testGetModTime() throws IOException {
+        long expectedModTime = tempFile.lastModified();
+        long actualModTime = blobStoreFile.getModTime();
+        assertEquals(expectedModTime, actualModTime, "The modification time should match the expected value.");
+    }
+}