Small updates on gcsync client (#829)

* Initial push for gcsync command line tool

* Remove unintentional changes

* Some updates

* Address pr comments

* Reimplement gcsbytesource with a cached inputstream

* remove unnecessary lines in build.gradle

* fix a typo

* Remember to close the outputstream

* The correct change for last committ

* Fix checksumgeneratortest

* These lines shouldn't be commented out

* Address some comments

* Add a timeout for cloud run tasks

* Use bytes instead of string to store the 256bit strong check sum which
would be more storage efficient

* set max timeout to 60 mins

Author: tonyxiaowei

dbsync/common/src/main/java/com/google/edwmigration/dbsync/common/ChecksumGenerator.java

@@ -5,9 +5,9 @@
 import com.google.common.hash.HashCode;
 import com.google.common.io.ByteSource;
 import com.google.common.primitives.Ints;
+import com.google.protobuf.ByteString;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.function.Consumer;
 import org.checkerframework.checker.index.qual.NonNegative;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,7 +48,8 @@ public void generate(ChecksumConsumer<IOException> out, ByteSource in) throws IO
         // Yes, this masks the instance variable.
         int blockSize = rollingChecksum.getBlockSize();
         if (DEBUG) {
-          logger.info("Generating checksums for [{} .. +{}] in {} bytes", offset, blockSize, dataSize);
+          logger.info("Generating checksums for [{} .. +{}] in {} bytes", offset, blockSize,
+              dataSize);
         }
 
         // If someone changes the size of the ByteSource underneath us, this might throw EOFException.
@@ -70,7 +71,7 @@ public void generate(ChecksumConsumer<IOException> out, ByteSource in) throws IO
             .setBlockOffset(offset)
             .setBlockLength(blockSize)
             .setWeakChecksum(weakHashCode)
-            .setStrongChecksum(strongHashCode.toString())
+            .setStrongChecksum(ByteString.copyFrom(strongHashCode.asBytes()))
             .build();
         out.accept(c);
       }

dbsync/common/src/main/java/com/google/edwmigration/dbsync/common/InstructionGenerator.java

@@ -13,6 +13,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.function.Consumer;
@@ -74,9 +75,9 @@ public void generate(InstructionConsumer<? extends IOException> out, ByteSource
           // And the wisdom to know the difference.
           MATCH:
           if (cc != null) {
-            String strongHashCode = rollingChecksum.getStrongHashCode().toString();
+            byte[] strongHashCode = rollingChecksum.getStrongHashCode().asBytes();
             for (Checksum c : cc) {
-              if (c.getStrongChecksum().equals(strongHashCode)) {
+              if (Arrays.equals(c.getStrongChecksum().toByteArray(), strongHashCode)) {
                 if (literalBufferLength > 0) {
                   if (DEBUG) {
                     logger.debug("Emitting pre-match literal");

dbsync/common/src/main/proto/dbsync.proto

@@ -13,7 +13,7 @@ message Checksum {
 	uint64 blockOffset = 1;
 	uint32 blockLength = 2;
 	int32 weakChecksum = 3;
-	string strongChecksum = 4;
+	bytes strongChecksum = 4;
 }
 
 message Instruction  {

dbsync/common/src/test/java/com/google/edwmigration/dbsync/common/ChecksumGeneratorTest.java

@@ -16,6 +16,7 @@
 import org.slf4j.LoggerFactory;
 
 public class ChecksumGeneratorTest {
+
   @SuppressWarnings("unused")
   private static final Logger logger = LoggerFactory.getLogger(ChecksumGeneratorTest.class);
 
@@ -40,10 +41,11 @@ public void testChecksumGenerator() throws Exception {
     assertEquals(blockCount, checksums.size());
     for (int i = 0; i < blockCount; i++) {
       Checksum c = checksums.get(i);
-      assertEquals(i * blockSize, c.getBlockOffset(),"Bad offset in " + c);
+      assertEquals(i * blockSize, c.getBlockOffset(), "Bad offset in " + c);
       ByteSource block = data.slice(c.getBlockOffset(), c.getBlockLength());
       HashCode strongHashCode = block.hash(RollingChecksumImpl.STRONG_HASH_FUNCTION);
-      assertEquals(strongHashCode.toString(), c.getStrongChecksum(), "Bad hash code in " + c);
+      assertArrayEquals(strongHashCode.asBytes(), c.getStrongChecksum().toByteArray(),
+          "Bad hash code in " + c);
     }
   }
 }

dbsync/gcsync/src/main/java/com/google/edwmigration/dbsync/gcsync/Constants.java

@@ -1,5 +1,8 @@
 package com.google.edwmigration.dbsync.gcsync;
 
+import com.google.protobuf.Duration;
+import com.google.protobuf.util.Durations;
+
 public class Constants {
 
   public static final String FILES_TO_RSYNC_FILE_NAME = "filesToRsync.txt";
@@ -20,4 +23,6 @@ public class Constants {
   public static final String GENERATE_CHECK_SUM_MAIN = GenerateCheckSumMain.class.getName();
 
   public static final String RECONSTRUCT_FILE_MAIN = ReconstructFilesMain.class.getName();
+
+  public static final Duration CLOUD_RUN_TIMEOUT = Durations.fromMinutes(60);
 }

dbsync/gcsync/src/main/java/com/google/edwmigration/dbsync/gcsync/GcsyncClient.java

@@ -221,10 +221,16 @@ private void uploadRemainingFiles() throws URISyntaxException, IOException {
   private void runCloudRunJob(String command, String project)
       throws ExecutionException, InterruptedException {
     Job job = Job.newBuilder().setTemplate(ExecutionTemplate.newBuilder().setTemplate(
-        TaskTemplate.newBuilder().addContainers(
-            Container.newBuilder().setImage("docker.io/getbamba/google-cloud-sdk-java:latest")
-                .addCommand("/bin/sh")
-                .addAllArgs(Arrays.asList("-c", command)).build()).build()).build()).build();
+                TaskTemplate.newBuilder().
+                    setTimeout(Constants.CLOUD_RUN_TIMEOUT).
+                    addContainers(Container.
+                        newBuilder().
+                        setImage("docker.io/getbamba/google-cloud-sdk-java:latest")
+                        .addCommand("/bin/sh")
+                        .addAllArgs(Arrays.asList("-c", command)).build())
+                    .build())
+            .build())
+        .build();
 
     String jobId = String.format("gcsync-cloudrun-%s", UUID.randomUUID());
     CreateJobRequest createJobRequest = CreateJobRequest.newBuilder()

dbsync/gcsync/src/main/java/com/google/edwmigration/dbsync/gcsync/GenerateCheckSumMain.java

@@ -7,7 +7,6 @@
 import com.google.edwmigration.dbsync.common.ChecksumGenerator;
 import com.google.edwmigration.dbsync.common.DefaultArguments;
 import com.google.edwmigration.dbsync.storage.gcs.GcsStorage;
-import java.io.IOException;
 import java.io.OutputStream;
 import java.net.URI;
 import java.util.List;