test(bigtable): fix integration tests for prod (#9464)

google/cloud/bigtable/admin/integration_tests/instance_admin_integration_test.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/bigtable/admin/bigtable_instance_admin_client.h"
+#include "google/cloud/bigtable/app_profile_config.h"
 #include "google/cloud/bigtable/iam_binding.h"
 #include "google/cloud/bigtable/iam_policy.h"
 #include "google/cloud/bigtable/resource_names.h"
@@ -222,11 +223,15 @@ TEST_F(InstanceAdminIntegrationTest, CreateListGetDeleteAppProfile) {
   EXPECT_THAT(*profiles, Not(Contains(name_1)));
   EXPECT_THAT(*profiles, Not(Contains(name_2)));
 
-  auto profile_1 = client_.CreateAppProfile(instance_name, id_1, {});
+  auto ap_1 = bigtable::AppProfileConfig::MultiClusterUseAny(id_1).as_proto();
+  ap_1.set_parent(instance_name);
+  auto profile_1 = client_.CreateAppProfile(ap_1);
   ASSERT_STATUS_OK(profile_1);
   EXPECT_EQ(profile_1->name(), name_1);
 
-  auto profile_2 = client_.CreateAppProfile(instance_name, id_2, {});
+  auto ap_2 = bigtable::AppProfileConfig::MultiClusterUseAny(id_2).as_proto();
+  ap_2.set_parent(instance_name);
+  auto profile_2 = client_.CreateAppProfile(ap_2);
   ASSERT_STATUS_OK(profile_2);
   EXPECT_EQ(profile_2->name(), name_2);
 
@@ -394,10 +399,10 @@ TEST_F(InstanceAdminIntegrationTest, SetGetTestIamAPIsTest) {
   auto iam_policy = bigtable::IamPolicy({bigtable::IamBinding(
       "roles/bigtable.reader", {"serviceAccount:" + service_account_})});
 
-  auto initial_policy = client_.SetIamPolicy(instance_id, iam_policy);
+  auto initial_policy = client_.SetIamPolicy(instance_name, iam_policy);
   ASSERT_STATUS_OK(initial_policy);
 
-  auto fetched_policy = client_.GetIamPolicy(instance_id);
+  auto fetched_policy = client_.GetIamPolicy(instance_name);
   ASSERT_STATUS_OK(fetched_policy);
 
   EXPECT_EQ(initial_policy->version(), fetched_policy->version());

google/cloud/bigtable/admin/integration_tests/table_admin_integration_test.cc

@@ -16,6 +16,8 @@
 #include "google/cloud/bigtable/admin/bigtable_table_admin_client.h"
 #include "google/cloud/bigtable/resource_names.h"
 #include "google/cloud/bigtable/testing/table_integration_test.h"
+#include "google/cloud/bigtable/wait_for_consistency.h"
+#include "google/cloud/internal/background_threads_impl.h"
 #include "google/cloud/internal/backoff_policy.h"
 #include "google/cloud/internal/getenv.h"
 #include "google/cloud/internal/random.h"
@@ -303,8 +305,10 @@ TEST_F(TableAdminIntegrationTest, WaitForConsistencyCheck) {
 
   // We need to mutate the data in the table and then wait for those mutations
   // to propagate to both clusters. First create a `bigtable::Table` object.
+  internal::AutomaticallyCreatedBackgroundThreads background;
   auto table = bigtable::Table(
-      bigtable::MakeDataConnection(),
+      bigtable::MakeDataConnection(
+          Options{}.set<GrpcCompletionQueueOption>(background.cq())),
       bigtable::TableResource(project_id(), id, random_table_id));
 
   // Insert some cells into the table.
@@ -325,39 +329,14 @@ TEST_F(TableAdminIntegrationTest, WaitForConsistencyCheck) {
   auto consistency_token =
       std::move(*consistency_token_resp->mutable_consistency_token());
 
-  // A retry loop that checks if the data replication has completed.
-  auto wait_for_consistency = [&]() -> Status {
-    auto retry = google::cloud::internal::LimitedTimeRetryPolicy<
-        bigtable::internal::SafeGrpcRetry>(std::chrono::minutes(2));
-    auto backoff = google::cloud::internal::ExponentialBackoffPolicy(
-        std::chrono::milliseconds(5), std::chrono::minutes(2), 2.0);
-
-    Status status;
-    do {
-      auto result = client_.CheckConsistency(table_name, consistency_token);
-      if (result.ok()) {
-        // Consistency has been achieved. Break the loop.
-        if (result->consistent()) return Status();
-        status = Status(StatusCode::kUnavailable, "Not consistent yet");
-      } else {
-        status = std::move(result).status();
-      }
-      if (!retry.OnFailure(status)) {
-        // The retry policy is exhausted or the error is not retryable.
-        return status;
-      }
-      std::this_thread::sleep_for(backoff.OnCompletion());
-    } while (!retry.IsExhausted());
-    return status;
-  };
-
-  // Verify that our clusters are eventually consistent.
-  auto result = wait_for_consistency();
-  ASSERT_STATUS_OK(result);
+  // Verify that our clusters are eventually consistent. This calls
+  // `AsyncCheckConsistency` under the hood.
+  auto result = AsyncWaitForConsistency(background.cq(), client_,
+                                        table.table_name(), consistency_token);
+  ASSERT_STATUS_OK(result.get());
 
-  // Make an asynchronous call, just to test all functions.
-  auto resp =
-      client_.AsyncCheckConsistency(table_name, consistency_token).get();
+  // Make a synchronous call, just to test all functions.
+  auto resp = client_.CheckConsistency(table_name, consistency_token);
   ASSERT_STATUS_OK(resp);
   EXPECT_TRUE(resp->consistent());