feat: allow restore backup to different instance (#515)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [x] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/java-bigtable/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [x] Ensure the tests and linter pass
- [x] Code coverage does not decrease (if any source code was changed)
- [x] Appropriate docs were updated (if necessary)

Fixes #789
kokoro:force-run

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/admin/v2/models/RestoreTableRequest.java

@@ -20,24 +20,46 @@
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 /** Fluent wrapper for {@link com.google.bigtable.admin.v2.RestoreTableRequest} */
 public final class RestoreTableRequest {
   private final com.google.bigtable.admin.v2.RestoreTableRequest.Builder requestBuilder =
       com.google.bigtable.admin.v2.RestoreTableRequest.newBuilder();
-  private final String backupId;
-  private final String clusterId;
+  private final String sourceBackupId;
+  private final String sourceClusterId;
+  private final String sourceInstanceId;
 
-  public static RestoreTableRequest of(String clusterId, String backupId) {
-    RestoreTableRequest request = new RestoreTableRequest(clusterId, backupId);
+  /**
+   * Create a {@link RestoreTableRequest} object. It assumes the source backup locates in the same
+   * instance as the destination table. To restore a table from a backup in another instance, use
+   * {@link #of(String, String, String) of} method.
+   */
+  public static RestoreTableRequest of(String sourceClusterId, String sourceBackupId) {
+    RestoreTableRequest request = new RestoreTableRequest(null, sourceClusterId, sourceBackupId);
     return request;
   }
 
-  private RestoreTableRequest(String clusterId, String backupId) {
-    Preconditions.checkNotNull(clusterId);
-    Preconditions.checkNotNull(backupId);
-    this.backupId = backupId;
-    this.clusterId = clusterId;
+  /**
+   * Create a {@link RestoreTableRequest} object. The source backup could locate in a the same or a
+   * different instance.
+   */
+  public static RestoreTableRequest of(
+      String sourceInstanceId, String sourceClusterId, String sourceBackupId) {
+    RestoreTableRequest request =
+        new RestoreTableRequest(sourceInstanceId, sourceClusterId, sourceBackupId);
+    return request;
+  }
+
+  private RestoreTableRequest(
+      @Nullable String sourceInstanceId,
+      @Nonnull String sourceClusterId,
+      @Nonnull String sourceBackupId) {
+    Preconditions.checkNotNull(sourceClusterId);
+    Preconditions.checkNotNull(sourceBackupId);
+    this.sourceBackupId = sourceBackupId;
+    this.sourceInstanceId = sourceInstanceId;
+    this.sourceClusterId = sourceClusterId;
   }
 
   public RestoreTableRequest setTableId(String tableId) {
@@ -56,13 +78,15 @@ public boolean equals(Object o) {
     }
     RestoreTableRequest that = (RestoreTableRequest) o;
     return Objects.equal(requestBuilder.getTableId(), that.requestBuilder.getTableId())
-        && Objects.equal(clusterId, that.clusterId)
-        && Objects.equal(backupId, that.backupId);
+        && Objects.equal(sourceInstanceId, that.sourceInstanceId)
+        && Objects.equal(sourceClusterId, that.sourceClusterId)
+        && Objects.equal(sourceBackupId, that.sourceBackupId);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hashCode(requestBuilder.getTableId(), clusterId, backupId);
+    return Objects.hashCode(
+        requestBuilder.getTableId(), sourceInstanceId, sourceClusterId, sourceBackupId);
   }
 
   @InternalApi
@@ -73,7 +97,12 @@ public com.google.bigtable.admin.v2.RestoreTableRequest toProto(
 
     return requestBuilder
         .setParent(NameUtil.formatInstanceName(projectId, instanceId))
-        .setBackup(NameUtil.formatBackupName(projectId, instanceId, clusterId, backupId))
+        .setBackup(
+            NameUtil.formatBackupName(
+                projectId,
+                sourceInstanceId == null ? instanceId : sourceInstanceId,
+                sourceClusterId,
+                sourceBackupId))
         .build();
   }
 }

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/admin/v2/it/BigtableBackupIT.java

@@ -103,26 +103,9 @@ public static void createClient()
     prefix = String.format("020%d", System.currentTimeMillis());
 
     tableAdmin = testEnvRule.env().getTableAdminClientForInstance(targetInstance);
-
-    testTable =
-        tableAdmin.createTable(
-            CreateTableRequest.of(generateId(TEST_TABLE_SUFFIX)).addFamily("cf1"));
-
-    // Populate test data.
     dataClient = testEnvRule.env().getDataClientForInstance(targetInstance);
-    byte[] rowBytes = new byte[1024];
-    Random random = new Random();
-    random.nextBytes(rowBytes);
 
-    List<ApiFuture<?>> futures = Lists.newArrayList();
-    for (int i = 0; i < 10; i++) {
-      ApiFuture<Void> future =
-          dataClient.mutateRowAsync(
-              RowMutation.create(testTable.getId(), "test-row-" + i)
-                  .setCell("cf1", ByteString.EMPTY, ByteString.copyFrom(rowBytes)));
-      futures.add(future);
-    }
-    ApiFutures.allAsList(futures).get(3, TimeUnit.MINUTES);
+    testTable = createAndPopulateTestTable(tableAdmin, dataClient);
   }
 
   @AfterClass
@@ -266,32 +249,89 @@ public void deleteBackupTest() throws InterruptedException {
   @Test
   public void restoreTableTest() throws InterruptedException, ExecutionException {
     String backupId = generateId("restore-" + TEST_BACKUP_SUFFIX);
-    String tableId = generateId("restored-table");
+    String restoredTableId = generateId("restored-table");
     tableAdmin.createBackup(createBackupRequest(backupId));
 
     // Wait 2 minutes so that the RestoreTable API will trigger an optimize restored
     // table operation.
     Thread.sleep(120 * 1000);
 
     try {
-      RestoreTableRequest req = RestoreTableRequest.of(targetCluster, backupId).setTableId(tableId);
+      RestoreTableRequest req =
+          RestoreTableRequest.of(targetCluster, backupId).setTableId(restoredTableId);
       RestoredTableResult result = tableAdmin.restoreTable(req);
       assertWithMessage("Incorrect restored table id")
           .that(result.getTable().getId())
-          .isEqualTo(tableId);
+          .isEqualTo(restoredTableId);
 
       if (result.getOptimizeRestoredTableOperationToken() != null) {
         // The assertion might be missing if the test is running against a HDD cluster or an
         // optimization is not necessary.
         tableAdmin.awaitOptimizeRestoredTable(result.getOptimizeRestoredTableOperationToken());
-        Table restoredTable = tableAdmin.getTable(tableId);
+        Table restoredTable = tableAdmin.getTable(restoredTableId);
         assertWithMessage("Incorrect restored table id")
             .that(restoredTable.getId())
-            .isEqualTo(tableId);
+            .isEqualTo(restoredTableId);
       }
     } finally {
       tableAdmin.deleteBackup(targetCluster, backupId);
-      tableAdmin.deleteTable(tableId);
+      tableAdmin.deleteTable(restoredTableId);
+    }
+  }
+
+  @Test
+  public void crossInstanceRestoreTest()
+      throws InterruptedException, IOException, ExecutionException, TimeoutException {
+    String backupId = generateId("cross-" + TEST_BACKUP_SUFFIX);
+    String restoredTableId = generateId("restored-table-2");
+
+    // Set up a new instance to test cross-instance restore. The source backup is stored in this
+    // instance.
+    String sourceInstance =
+        AbstractTestEnv.TEST_INSTANCE_PREFIX + "backup-" + Instant.now().getEpochSecond();
+    String sourceCluster = AbstractTestEnv.TEST_CLUSTER_PREFIX + Instant.now().getEpochSecond();
+    instanceAdmin.createInstance(
+        CreateInstanceRequest.of(sourceInstance)
+            .addCluster(sourceCluster, testEnvRule.env().getSecondaryZone(), 3, StorageType.SSD)
+            .setDisplayName("backups-source-test-instance")
+            .addLabel("state", "readytodelete")
+            .setType(Type.PRODUCTION));
+    BigtableTableAdminClient sourceTableAdmin =
+        testEnvRule.env().getTableAdminClientForInstance(sourceInstance);
+    Table sourceTable =
+        createAndPopulateTestTable(
+            sourceTableAdmin, testEnvRule.env().getDataClientForInstance(sourceInstance));
+    sourceTableAdmin.createBackup(
+        CreateBackupRequest.of(sourceCluster, backupId)
+            .setSourceTableId(sourceTable.getId())
+            .setExpireTime(Instant.now().plus(Duration.ofHours(6))));
+
+    // Wait 2 minutes so that the RestoreTable API will trigger an optimize restored
+    // table operation.
+    Thread.sleep(120 * 1000);
+
+    try {
+      RestoreTableRequest req =
+          RestoreTableRequest.of(sourceInstance, sourceCluster, backupId)
+              .setTableId(restoredTableId);
+      RestoredTableResult result = tableAdmin.restoreTable(req);
+      assertWithMessage("Incorrect restored table id")
+          .that(result.getTable().getId())
+          .isEqualTo(restoredTableId);
+      assertWithMessage("Incorrect instance id")
+          .that(result.getTable().getInstanceId())
+          .isEqualTo(targetInstance);
+
+      // The assertion might be missing if the test is running against a HDD cluster or an
+      // optimization is not necessary.
+      assertWithMessage("Empty OptimizeRestoredTable token")
+          .that(result.getOptimizeRestoredTableOperationToken())
+          .isNotNull();
+      tableAdmin.awaitOptimizeRestoredTable(result.getOptimizeRestoredTableOperationToken());
+      tableAdmin.getTable(restoredTableId);
+    } finally {
+      sourceTableAdmin.deleteBackup(sourceCluster, backupId);
+      instanceAdmin.deleteInstance(sourceInstance);
     }
   }
 
@@ -327,13 +367,37 @@ public void backupIamTest() throws InterruptedException {
     }
   }
 
-  private CreateBackupRequest createBackupRequest(String backupName) {
-    return CreateBackupRequest.of(targetCluster, backupName)
+  private CreateBackupRequest createBackupRequest(String backupId) {
+    return CreateBackupRequest.of(targetCluster, backupId)
         .setSourceTableId(testTable.getId())
         .setExpireTime(Instant.now().plus(Duration.ofDays(15)));
   }
 
   private static String generateId(String name) {
     return prefix + "-" + name;
   }
+
+  private static Table createAndPopulateTestTable(
+      BigtableTableAdminClient tableAdmin, BigtableDataClient dataClient)
+      throws InterruptedException, ExecutionException, TimeoutException {
+    Table testTable =
+        tableAdmin.createTable(
+            CreateTableRequest.of(generateId(TEST_TABLE_SUFFIX)).addFamily("cf1"));
+
+    // Populate test data.
+    byte[] rowBytes = new byte[1024];
+    Random random = new Random();
+    random.nextBytes(rowBytes);
+
+    List<ApiFuture<?>> futures = Lists.newArrayList();
+    for (int i = 0; i < 10; i++) {
+      ApiFuture<Void> future =
+          dataClient.mutateRowAsync(
+              RowMutation.create(testTable.getId(), "test-row-" + i)
+                  .setCell("cf1", ByteString.EMPTY, ByteString.copyFrom(rowBytes)));
+      futures.add(future);
+    }
+    ApiFutures.allAsList(futures).get(3, TimeUnit.MINUTES);
+    return testTable;
+  }
 }

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/admin/v2/models/RestoreTableRequestTest.java

@@ -30,6 +30,7 @@ public class RestoreTableRequestTest {
   private static final String PROJECT_ID = "my-project";
   private static final String INSTANCE_ID = "my-instance";
   private static final String CLUSTER_ID = "my-cluster";
+  private static final String SOURCE_INSTANCE_ID = "source-instance-id";
 
   @Test
   public void testToProto() {
@@ -45,6 +46,21 @@ public void testToProto() {
     assertThat(request.toProto(PROJECT_ID, INSTANCE_ID)).isEqualTo(requestProto);
   }
 
+  @Test
+  public void testToProtoCrossInstance() {
+    RestoreTableRequest request =
+        RestoreTableRequest.of(SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID);
+
+    com.google.bigtable.admin.v2.RestoreTableRequest requestProto =
+        com.google.bigtable.admin.v2.RestoreTableRequest.newBuilder()
+            .setParent(NameUtil.formatInstanceName(PROJECT_ID, INSTANCE_ID))
+            .setBackup(
+                NameUtil.formatBackupName(PROJECT_ID, SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID))
+            .setTableId(TABLE_ID)
+            .build();
+    assertThat(request.toProto(PROJECT_ID, INSTANCE_ID)).isEqualTo(requestProto);
+  }
+
   @Test
   public void testEquality() {
     RestoreTableRequest request =
@@ -56,6 +72,22 @@ public void testEquality() {
         .isNotEqualTo(RestoreTableRequest.of(CLUSTER_ID, BACKUP_ID).setTableId("another-table"));
   }
 
+  @Test
+  public void testEqualityCrossInstance() {
+    RestoreTableRequest request =
+        RestoreTableRequest.of(SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID);
+
+    assertThat(request)
+        .isEqualTo(
+            RestoreTableRequest.of(SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID));
+    assertThat(request)
+        .isNotEqualTo(RestoreTableRequest.of(CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID));
+    assertThat(request)
+        .isNotEqualTo(
+            RestoreTableRequest.of(SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID)
+                .setTableId("another-table"));
+  }
+
   @Test
   public void testHashCode() {
     RestoreTableRequest request =
@@ -66,4 +98,23 @@ public void testHashCode() {
         .isNotEqualTo(
             RestoreTableRequest.of(CLUSTER_ID, BACKUP_ID).setTableId("another-table").hashCode());
   }
+
+  @Test
+  public void testHashCodeCrossInstance() {
+    RestoreTableRequest request =
+        RestoreTableRequest.of(SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID);
+    assertThat(request.hashCode())
+        .isEqualTo(
+            RestoreTableRequest.of(SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID)
+                .setTableId(TABLE_ID)
+                .hashCode());
+    assertThat(request.hashCode())
+        .isNotEqualTo(
+            RestoreTableRequest.of(CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID).hashCode());
+    assertThat(request.hashCode())
+        .isNotEqualTo(
+            RestoreTableRequest.of(SOURCE_INSTANCE_ID, CLUSTER_ID, BACKUP_ID)
+                .setTableId("another-table")
+                .hashCode());
+  }
 }