impl(bigtable): save and restore Options across query plan refresh threads

Author: scotthart

google/cloud/bigtable/internal/data_connection_impl.cc

@@ -752,7 +752,8 @@ StatusOr<bigtable::PreparedQuery> DataConnectionImpl::PrepareQuery(
     return std::move(response).status();
   }
   auto const* func = __func__;
-  auto refresh_fn = [this, request, current, func]() mutable {
+  auto refresh_fn = [this, request, func]() mutable {
+    auto current = google::cloud::internal::SaveCurrentOptions();
     auto retry = retry_policy(*current);
     auto backoff = backoff_policy(*current);
     auto operation_context = operation_context_factory_->PrepareQuery(
@@ -834,7 +835,8 @@ future<StatusOr<bigtable::PreparedQuery>> DataConnectionImpl::AsyncPrepareQuery(
           return std::move(response).status();
         }
 
-        auto refresh_fn = [this, request, current, func]() mutable {
+        auto refresh_fn = [this, request, func]() mutable {
+          auto current = google::cloud::internal::SaveCurrentOptions();
           auto retry = retry_policy(*current);
           auto backoff = backoff_policy(*current);
           auto operation_context = operation_context_factory_->PrepareQuery(

google/cloud/bigtable/internal/query_plan.cc

@@ -50,8 +50,14 @@ void QueryPlan::ScheduleRefresh(std::unique_lock<std::mutex> const&) {
   std::weak_ptr<QueryPlan> plan = shared_from_this();
   refresh_timer_ =
       cq_.MakeDeadlineTimer(refresh_deadline)
-          .then([plan](future<StatusOr<std::chrono::system_clock::time_point>>
-                           result) {
+          .then([plan, current = internal::SaveCurrentOptions()](
+                    future<StatusOr<std::chrono::system_clock::time_point>>
+                        result) {
+            // Options are stored in a thread_local variable. When this timer
+            // expires and this lambda is executed we need to restore the
+            // Options that were saved in the capture group as a different
+            // thread may be used.
+            internal::OptionsSpan options_span(current);
             if (result.get().ok()) {
               if (auto p = plan.lock()) {
                 p->ExpiredRefresh();