Add staged package hashes (#38311)

Author: tarun-google

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/PackageUtil.java

@@ -438,6 +438,7 @@ public static PackageAttributes forFileToStage(
               .toString();
       destination.setLocation(resourcePath);
       destination.setName(dest);
+      destination.setSha256(hash);
       return new AutoValue_PackageUtil_PackageAttributes(
           file, null, destination, file.length(), hash);
     }
@@ -456,6 +457,7 @@ public static PackageAttributes forBytesToStage(
       DataflowPackage targetPackage = new DataflowPackage();
       targetPackage.setName(target);
       targetPackage.setLocation(resourcePath);
+      targetPackage.setSha256(hashCode.toString());
 
       return new AutoValue_PackageUtil_PackageAttributes(
           null, bytes, targetPackage, size, hashCode.toString());
@@ -465,6 +467,7 @@ public PackageAttributes withPackageName(String overridePackageName) {
       DataflowPackage newDestination = new DataflowPackage();
       newDestination.setName(overridePackageName);
       newDestination.setLocation(getDestination().getLocation());
+      newDestination.setSha256(getHash());
 
       return new AutoValue_PackageUtil_PackageAttributes(
           getSource(), getBytes(), newDestination, getSize(), getHash());

runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/PackageUtilTest.java

@@ -191,6 +191,8 @@ public void testFileWithExtensionPackageNamingAndSize() throws Exception {
     assertThat(target.getName(), endsWith(".txt"));
     assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
     assertThat(attr.getSize(), equalTo((long) contents.length()));
+    String expectedHash = Files.asByteSource(tmpFile).hash(Hashing.sha256()).toString();
+    assertThat(target.getSha256(), equalTo(expectedHash));
   }
 
   @Test
@@ -299,6 +301,8 @@ public void testPackageUploadWithFileSucceeds() throws Exception {
 
     assertThat(target.getName(), endsWith(".txt"));
     assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
+    String expectedHash = Files.asByteSource(tmpFile).hash(Hashing.sha256()).toString();
+    assertThat(target.getSha256(), equalTo(expectedHash));
     assertThat(
         new LineReader(Channels.newReader(pipe.source(), StandardCharsets.UTF_8.name())).readLine(),
         equalTo(contents));