Stabilize FileIOTest updated-file timestamp assertions

Author: bvolpato

sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java

@@ -232,9 +232,10 @@ public void testMatchAllDisallowEmptyNonWildcard() throws IOException {
   /** DoFn that copy test files from source to watch path. */
   private static class CopyFilesFn
       extends DoFn<KV<String, MatchResult.Metadata>, MatchResult.Metadata> {
-    public CopyFilesFn(Path sourcePath, Path watchPath) {
+    public CopyFilesFn(Path sourcePath, Path watchPath, long baseTimestampMillis) {
       this.sourcePathStr = sourcePath.toString();
       this.watchPathStr = watchPath.toString();
+      this.baseTimestampMillis = baseTimestampMillis;
     }
 
     @StateId("count")
@@ -257,10 +258,16 @@ public void processElement(ProcessContext context, @StateId("count") ValueState<
 
       if (0 == current) {
         Thread.sleep(100);
+        // Ensure overwrite updates get a distinct mtime even when COPY_ATTRIBUTES is enabled.
+        Files.setLastModifiedTime(
+            sourcePath.resolve("first"), FileTime.fromMillis(baseTimestampMillis + 2000));
         Files.copy(sourcePath.resolve("first"), watchPath.resolve("first"), updOptions);
         Files.copy(sourcePath.resolve("second"), watchPath.resolve("second"), cpOptions);
       } else if (1 == current) {
         Thread.sleep(100);
+        FileTime updateTime = FileTime.fromMillis(baseTimestampMillis + 4000);
+        Files.setLastModifiedTime(sourcePath.resolve("first"), updateTime);
+        Files.setLastModifiedTime(sourcePath.resolve("second"), updateTime);
         Files.copy(sourcePath.resolve("first"), watchPath.resolve("first"), updOptions);
         Files.copy(sourcePath.resolve("second"), watchPath.resolve("second"), updOptions);
         Files.copy(sourcePath.resolve("third"), watchPath.resolve("third"), cpOptions);
@@ -271,6 +278,7 @@ public void processElement(ProcessContext context, @StateId("count") ValueState<
     // Member variables need to be serializable.
     private final String sourcePathStr;
     private final String watchPathStr;
+    private final long baseTimestampMillis;
   }
 
   @Test
@@ -282,6 +290,12 @@ public void testMatchWatchForNewFiles() throws IOException, InterruptedException
     Files.write(sourcePath.resolve("first"), new byte[42]);
     Files.write(sourcePath.resolve("second"), new byte[37]);
     Files.write(sourcePath.resolve("third"), new byte[99]);
+    // Keep controlled mtimes in the past so updates are distinct without future timestamps.
+    long baseTimestampMillis = System.currentTimeMillis() - Duration.standardMinutes(1).getMillis();
+    FileTime baseTimestamp = FileTime.fromMillis(baseTimestampMillis);
+    Files.setLastModifiedTime(sourcePath.resolve("first"), baseTimestamp);
+    Files.setLastModifiedTime(sourcePath.resolve("second"), baseTimestamp);
+    Files.setLastModifiedTime(sourcePath.resolve("third"), baseTimestamp);
 
     // Create a "watch" directory that the pipeline will copy files into.
     final Path watchPath = tmpFolder.getRoot().toPath().resolve("watch");
@@ -336,7 +350,7 @@ public void testMatchWatchForNewFiles() throws IOException, InterruptedException
                             TypeDescriptors.strings(),
                             TypeDescriptor.of(MatchResult.Metadata.class)))
                     .via((metadata) -> KV.of("dumb key", metadata)))
-            .apply(ParDo.of(new CopyFilesFn(sourcePath, watchPath)));
+            .apply(ParDo.of(new CopyFilesFn(sourcePath, watchPath, baseTimestampMillis)));
 
     assertEquals(PCollection.IsBounded.UNBOUNDED, matchMetadata.isBounded());
     assertEquals(PCollection.IsBounded.UNBOUNDED, matchAllMetadata.isBounded());