impl(v3): add bespoke WaitForConsistency to bigtable table admin

generator/generator_config.textproto

@@ -588,6 +588,13 @@ service {
     {rpc_name: "BigtableTableAdmin.CheckConsistency", idempotency: IDEMPOTENT}
   ]
   omit_repo_metadata: true
+  bespoke_methods : [
+    {
+    name: "WaitForConsistency",
+    return_type: "future<StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>",
+    parameters: "(google::bigtable::admin::v2::CheckConsistencyRequest const& request, Options opts = {})"
+    }
+  ]
 }
 
 # Billing

google/cloud/bigtable/CMakeLists.txt

@@ -89,6 +89,7 @@ add_library(
     admin/internal/bigtable_table_admin_auth_decorator.h
     admin/internal/bigtable_table_admin_connection_impl.cc
     admin/internal/bigtable_table_admin_connection_impl.h
+    admin/internal/bigtable_table_admin_connection_impl_bespoke.cc
     admin/internal/bigtable_table_admin_logging_decorator.cc
     admin/internal/bigtable_table_admin_logging_decorator.h
     admin/internal/bigtable_table_admin_metadata_decorator.cc

google/cloud/bigtable/admin/bigtable_table_admin_client.cc

@@ -821,6 +821,14 @@ BigtableTableAdminClient::AsyncCheckConsistency(
   return connection_->AsyncCheckConsistency(request);
 }
 
+future<StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
+BigtableTableAdminClient::WaitForConsistency(
+    google::bigtable::admin::v2::CheckConsistencyRequest const& request,
+    Options opts) {
+  internal::OptionsSpan span(internal::MergeOptions(std::move(opts), options_));
+  return connection_->WaitForConsistency(request);
+}
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable_admin
 }  // namespace cloud

google/cloud/bigtable/admin/bigtable_table_admin_client.h

@@ -2581,6 +2581,40 @@ class BigtableTableAdminClient {
       google::bigtable::admin::v2::CheckConsistencyRequest const& request,
       Options opts = {});
 
+  // clang-format off
+  ///
+  /// Polls a table until it is consistent or the RetryPolicy is exhausted based
+  /// on a consistency token, that is, if replication has caught up based on the
+  /// provided conditions specified in the token and the check request.
+  ///
+  /// @param request Unary RPCs, such as the one wrapped by this
+  ///     function, receive a single `request` proto message which includes all
+  ///     the inputs for the RPC. In this case, the proto message is a
+  ///     [google.bigtable.admin.v2.CheckConsistencyRequest].
+  ///     Proto messages are converted to C++ classes by Protobuf, using the
+  ///     [Protobuf mapping rules].
+  /// @param opts Optional. Override the class-level options, such as retry and
+  ///     backoff policies.
+  /// @return the result of the RPC. The response message type
+  ///     ([google.bigtable.admin.v2.CheckConsistencyResponse])
+  ///     is mapped to a C++ class using the [Protobuf mapping rules].
+  ///     If the request fails, the [`StatusOr`] contains the error details.
+  ///
+  /// [Protobuf mapping rules]: https://protobuf.dev/reference/cpp/cpp-generated/
+  /// [input iterator requirements]: https://en.cppreference.com/w/cpp/named_req/InputIterator
+  /// [`std::string`]: https://en.cppreference.com/w/cpp/string/basic_string
+  /// [`future`]: @ref google::cloud::future
+  /// [`StatusOr`]: @ref google::cloud::StatusOr
+  /// [`Status`]: @ref google::cloud::Status
+  /// [google.bigtable.admin.v2.CheckConsistencyRequest]: @googleapis_reference_link{google/bigtable/admin/v2/bigtable_table_admin.proto#L909}
+  /// [google.bigtable.admin.v2.CheckConsistencyResponse]: @googleapis_reference_link{google/bigtable/admin/v2/bigtable_table_admin.proto#L948}
+  ///
+  // clang-format on
+  future<StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
+  WaitForConsistency(
+      google::bigtable::admin::v2::CheckConsistencyRequest const& request,
+      Options opts = {});
+
  private:
   std::shared_ptr<BigtableTableAdminConnection> connection_;
   Options options_;

google/cloud/bigtable/admin/bigtable_table_admin_connection.cc

@@ -382,6 +382,14 @@ BigtableTableAdminConnection::AsyncCheckConsistency(
       Status(StatusCode::kUnimplemented, "not implemented"));
 }
 
+future<StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
+BigtableTableAdminConnection::WaitForConsistency(
+    google::bigtable::admin::v2::CheckConsistencyRequest const&) {
+  return google::cloud::make_ready_future<
+      StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>(
+      Status(StatusCode::kUnimplemented, "not implemented"));
+}
+
 std::shared_ptr<BigtableTableAdminConnection> MakeBigtableTableAdminConnection(
     Options options) {
   internal::CheckExpectedOptions<CommonOptionList, GrpcOptionList,

google/cloud/bigtable/admin/bigtable_table_admin_connection.h

@@ -359,6 +359,11 @@ class BigtableTableAdminConnection {
       StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
   AsyncCheckConsistency(
       google::bigtable::admin::v2::CheckConsistencyRequest const& request);
+
+  virtual future<
+      StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
+  WaitForConsistency(
+      google::bigtable::admin::v2::CheckConsistencyRequest const& request);
 };
 
 /**

google/cloud/bigtable/admin/internal/bigtable_table_admin_connection_impl.h

@@ -234,6 +234,10 @@ class BigtableTableAdminConnectionImpl
       google::bigtable::admin::v2::CheckConsistencyRequest const& request)
       override;
 
+  future<StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
+  WaitForConsistency(google::bigtable::admin::v2::CheckConsistencyRequest const&
+                         request) override;
+
  private:
   std::unique_ptr<google::cloud::BackgroundThreads> background_;
   std::shared_ptr<bigtable_admin_internal::BigtableTableAdminStub> stub_;

google/cloud/bigtable/admin/internal/bigtable_table_admin_connection_impl_bespoke.cc

@@ -0,0 +1,76 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/bigtable/admin/internal/bigtable_table_admin_connection_impl.h"
+#include "google/cloud/common_options.h"
+#include "google/cloud/internal/async_retry_loop.h"
+#include <memory>
+#include <utility>
+
+namespace google {
+namespace cloud {
+namespace bigtable_admin_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace {
+
+std::unique_ptr<bigtable_admin::BigtableTableAdminRetryPolicy> retry_policy(
+    Options const& options) {
+  return options.get<bigtable_admin::BigtableTableAdminRetryPolicyOption>()
+      ->clone();
+}
+
+std::unique_ptr<BackoffPolicy> backoff_policy(Options const& options) {
+  return options.get<bigtable_admin::BigtableTableAdminBackoffPolicyOption>()
+      ->clone();
+}
+
+std::unique_ptr<bigtable_admin::BigtableTableAdminConnectionIdempotencyPolicy>
+idempotency_policy(Options const& options) {
+  return options
+      .get<
+          bigtable_admin::BigtableTableAdminConnectionIdempotencyPolicyOption>()
+      ->clone();
+}
+
+}  // namespace
+
+future<StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
+BigtableTableAdminConnectionImpl::WaitForConsistency(
+    google::bigtable::admin::v2::CheckConsistencyRequest const& request) {
+  auto current = google::cloud::internal::SaveCurrentOptions();
+  auto request_copy = request;
+  auto const idempotent =
+      idempotency_policy(*current)->CheckConsistency(request_copy);
+  auto retry = retry_policy(*current);
+  auto backoff = backoff_policy(*current);
+  auto attempt_predicate =
+      [](StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse> const&
+             r) { return r.ok() && r->consistent(); };
+  return google::cloud::internal::AsyncRetryLoop(
+      std::move(retry), std::move(backoff), idempotent, background_->cq(),
+      [stub = stub_](
+          CompletionQueue& cq, std::shared_ptr<grpc::ClientContext> context,
+          google::cloud::internal::ImmutableOptions options,
+          google::bigtable::admin::v2::CheckConsistencyRequest const& request) {
+        return stub->AsyncCheckConsistency(cq, std::move(context),
+                                           std::move(options), request);
+      },
+      std::move(current), std::move(request_copy), __func__,
+      std::move(attempt_predicate));
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace bigtable_admin_internal
+}  // namespace cloud
+}  // namespace google

google/cloud/bigtable/examples/table_admin_snippets.cc

@@ -633,30 +633,24 @@ void WaitForConsistencyCheck(
   using ::google::cloud::future;
   using ::google::cloud::Status;
   using ::google::cloud::StatusOr;
-  [](cbta::BigtableTableAdminClient admin, std::string const& project_id,
+  [](cbta::BigtableTableAdminClient, std::string const& project_id,
      std::string const& instance_id, std::string const& table_id) {
+    auto client = cbta::BigtableTableAdminClient(
+        cbta::MakeBigtableTableAdminConnection());
     std::string table_name = cbt::TableName(project_id, instance_id, table_id);
     StatusOr<google::bigtable::admin::v2::GenerateConsistencyTokenResponse>
-        consistency_token = admin.GenerateConsistencyToken(table_name);
+        consistency_token = client.GenerateConsistencyToken(table_name);
     if (!consistency_token) {
       throw std::move(consistency_token).status();
     }
-    // Start a thread to perform the background work.
-    CompletionQueue cq;
-    std::thread cq_runner([&cq] { cq.Run(); });
-
-    std::string token = consistency_token->consistency_token();
-    future<Status> consistent_future =
-        cbta::AsyncWaitForConsistency(cq, admin, table_name, token);
-
-    // Simplify the example by blocking until the operation is done.
-    Status status = consistent_future.get();
-    if (!status.ok()) throw std::runtime_error(status.message());
+    auto token = consistency_token->consistency_token();
+    google::bigtable::admin::v2::CheckConsistencyRequest wait_request;
+    wait_request.set_name(table_name);
+    wait_request.set_consistency_token(token);
+    auto consistency_future = client.WaitForConsistency(wait_request);
+    auto consistency = consistency_future.get();
+    if (!consistency) throw std::runtime_error(consistency.status().message());
     std::cout << "Table is consistent with token " << token << "\n";
-
-    // Shutdown the work queue and join the background thread
-    cq.Shutdown();
-    cq_runner.join();
   }
   //! [wait for consistency check]
   (std::move(admin), argv.at(0), argv.at(1), argv.at(2));

google/cloud/bigtable/google_cloud_cpp_bigtable.bzl

@@ -163,6 +163,7 @@ google_cloud_cpp_bigtable_srcs = [
     "admin/internal/bigtable_instance_admin_tracing_stub.cc",
     "admin/internal/bigtable_table_admin_auth_decorator.cc",
     "admin/internal/bigtable_table_admin_connection_impl.cc",
+    "admin/internal/bigtable_table_admin_connection_impl_bespoke.cc",
     "admin/internal/bigtable_table_admin_logging_decorator.cc",
     "admin/internal/bigtable_table_admin_metadata_decorator.cc",
     "admin/internal/bigtable_table_admin_option_defaults.cc",

google/cloud/bigtable/tests/admin_integration_test.cc

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "google/cloud/bigtable/admin/bigtable_instance_admin_client.h"
+#include "google/cloud/bigtable/admin/bigtable_table_admin_client.h"
 #include "google/cloud/bigtable/instance_admin.h"
 #include "google/cloud/bigtable/table_admin.h"
 #include "google/cloud/bigtable/testing/table_integration_test.h"
@@ -221,11 +223,13 @@ TEST_F(AdminIntegrationTest, WaitForConsistencyCheck) {
 
   // Create a bigtable::InstanceAdmin and a bigtable::TableAdmin to create the
   // new instance and the new table.
-  auto instance_admin_client = bigtable::MakeInstanceAdminClient(project_id());
-  bigtable::InstanceAdmin instance_admin(instance_admin_client);
-
-  auto admin_client = bigtable::MakeAdminClient(project_id());
-  bigtable::TableAdmin table_admin(admin_client, id);
+  auto instance_admin =
+      std::make_unique<bigtable_admin::BigtableInstanceAdminClient>(
+          bigtable_admin::MakeBigtableInstanceAdminConnection());
+  auto table_admin_connection =
+      bigtable_admin::MakeBigtableTableAdminConnection();
+  auto table_admin = std::make_unique<bigtable_admin::BigtableTableAdminClient>(
+      table_admin_connection);
 
   // The instance configuration is involved, it needs two clusters, which must
   // be production clusters (and therefore have at least 3 nodes each), and
@@ -243,7 +247,9 @@ TEST_F(AdminIntegrationTest, WaitForConsistencyCheck) {
       {{id + "-c1", cluster_config_1}, {id + "-c2", cluster_config_2}});
 
   // Create the new instance.
-  auto instance = instance_admin.CreateInstance(config).get();
+  auto create_request = config.as_proto();
+  create_request.set_parent(Project(project_id()).FullName());
+  auto instance = instance_admin->CreateInstance(create_request).get();
   ASSERT_STATUS_OK(instance);
 
   // The table is going to be very simple, just one column family.
@@ -252,7 +258,10 @@ TEST_F(AdminIntegrationTest, WaitForConsistencyCheck) {
       {{family, bigtable::GcRule::MaxNumVersions(10)}}, {});
 
   // Create the new table.
-  auto table_created = table_admin.CreateTable(random_table_id, table_config);
+  auto request = std::move(table_config).as_proto();
+  request.set_parent(InstanceName(project_id(), id));
+  request.set_table_id(random_table_id);
+  auto table_created = table_admin->CreateTable(request);
   ASSERT_STATUS_OK(table_created);
 
   // We need to mutate the data in the table and then wait for those mutations
@@ -273,20 +282,24 @@ TEST_F(AdminIntegrationTest, WaitForConsistencyCheck) {
 
   // Create a consistency token after modifying the table.
   auto consistency_token =
-      table_admin.GenerateConsistencyToken(random_table_id);
+      table_admin->GenerateConsistencyToken(table_created->name());
   ASSERT_STATUS_OK(consistency_token);
 
   // Wait until all the mutations before the `consistency_token` have propagated
   // everywhere.
-  google::cloud::future<google::cloud::StatusOr<bigtable::Consistency>> result =
-      table_admin.WaitForConsistency(random_table_id, *consistency_token);
-  auto is_consistent = result.get();
+  google::bigtable::admin::v2::CheckConsistencyRequest wait_request;
+  wait_request.set_name(table_created->name());
+  wait_request.set_consistency_token(consistency_token->consistency_token());
+  future<StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>>
+      result = table_admin->WaitForConsistency(wait_request);
+  StatusOr<google::bigtable::admin::v2::CheckConsistencyResponse>
+      is_consistent = result.get();
   ASSERT_STATUS_OK(is_consistent);
-  EXPECT_EQ(bigtable::Consistency::kConsistent, *is_consistent);
+  EXPECT_TRUE(is_consistent->consistent());
 
   // Cleanup the table and the instance.
-  EXPECT_STATUS_OK(table_admin.DeleteTable(random_table_id));
-  EXPECT_STATUS_OK(instance_admin.DeleteInstance(id));
+  EXPECT_STATUS_OK(table_admin->DeleteTable(table_created->name()));
+  EXPECT_STATUS_OK(instance_admin->DeleteInstance(instance->name()));
 }
 
 /// @test Verify rpc logging for `bigtable::TableAdmin`