diff --git a/core/client/hdfs/src/main/java/alluxio/hadoop/LocalCacheFileSystem.java b/core/client/hdfs/src/main/java/alluxio/hadoop/LocalCacheFileSystem.java
index 3781a3b1cfe2..88e6002a5ffa 100644
--- a/core/client/hdfs/src/main/java/alluxio/hadoop/LocalCacheFileSystem.java
+++ b/core/client/hdfs/src/main/java/alluxio/hadoop/LocalCacheFileSystem.java
@@ -21,6 +21,7 @@
 import alluxio.client.file.cache.LocalCacheFileInStream;
 import alluxio.client.file.cache.filter.CacheFilter;
 import alluxio.conf.AlluxioConfiguration;
+import alluxio.conf.PropertyKey;
 import alluxio.metrics.MetricsConfig;
 import alluxio.metrics.MetricsSystem;
 import alluxio.wire.FileInfo;
@@ -130,8 +131,15 @@ public FSDataInputStream open(Path path, int bufferSize) throws IOException {
         .setGroup(externalFileStatus.getGroup());
     // FilePath is a unique identifier for a file, however it can be a long string
     // hence using md5 hash of the file path as the identifier in the cache.
-    CacheContext context = CacheContext.defaults().setCacheIdentifier(
-        md5().hashString(externalFileStatus.getPath().toString(), UTF_8).toString());
+    String cacheIdentifier;
+    if (mAlluxioConf.getBoolean(PropertyKey.USER_CLIENT_CACHE_IDENTIFIER_INCLUDE_MTIME)) {
+      // include mtime to avoid consistency issues if the file may update
+      cacheIdentifier = md5().hashString(externalFileStatus.getPath().toString()
+          + externalFileStatus.getModificationTime(), UTF_8).toString();
+    } else {
+      cacheIdentifier = md5().hashString(externalFileStatus.getPath().toString(), UTF_8).toString();
+    }
+    CacheContext context = CacheContext.defaults().setCacheIdentifier(cacheIdentifier);
     URIStatus status = new URIStatus(info, context);
     return open(status, bufferSize);
   }
diff --git a/core/common/src/main/java/alluxio/conf/PropertyKey.java b/core/common/src/main/java/alluxio/conf/PropertyKey.java
index d5c6a5e58fa7..e52e2c41affd 100755
--- a/core/common/src/main/java/alluxio/conf/PropertyKey.java
+++ b/core/common/src/main/java/alluxio/conf/PropertyKey.java
@@ -6229,6 +6229,14 @@ public String toString() {
           .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
           .setScope(Scope.CLIENT)
           .build();
+  public static final PropertyKey USER_CLIENT_CACHE_IDENTIFIER_INCLUDE_MTIME =
+      booleanBuilder(Name.USER_CLIENT_CACHE_IDENTIFIER_INCLUDE_MTIME)
+          .setDefaultValue(false)
+          .setDescription("If this is enabled, client-side cache will include modification time "
+              + "while calculating the identifier of a file.")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
+          .setScope(Scope.CLIENT)
+          .build();
 
   public static final PropertyKey USER_CLIENT_REPORT_VERSION_ENABLED =
       booleanBuilder(Name.USER_CLIENT_REPORT_VERSION_ENABLED)
@@ -8879,6 +8887,8 @@ public static final class Name {
         "alluxio.user.client.cache.timeout.duration";
     public static final String USER_CLIENT_CACHE_TIMEOUT_THREADS =
         "alluxio.user.client.cache.timeout.threads";
+    public static final String USER_CLIENT_CACHE_IDENTIFIER_INCLUDE_MTIME =
+        "alluxio.user.client.cache.include.mtime";
     public static final String USER_CLIENT_REPORT_VERSION_ENABLED =
         "alluxio.user.client.report.version.enabled";
     public static final String USER_CONF_CLUSTER_DEFAULT_ENABLED =