chore: merge main into generate-libraries-main

Author: cloud-java-bot

google-cloud-bigtable-emulator-core/src/main/java/com/google/cloud/bigtable/emulator/core/EmulatorController.java

@@ -129,8 +129,8 @@ public synchronized void start(int port)
         throw e;
       }
     }
-    Thread stdoutThread = pipeStreamToLog(process.getInputStream(), Level.INFO);
-    Thread stderrThread = pipeStreamToLog(process.getErrorStream(), Level.WARNING);
+    stdoutThread = pipeStreamToLog(process.getInputStream(), Level.INFO);
+    stderrThread = pipeStreamToLog(process.getErrorStream(), Level.WARNING);
     isStopped = false;
 
     shutdownHook =

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/PreparedStatementImpl.java

@@ -154,7 +154,7 @@ ApiFuture<PrepareResponse> getFreshPlan() {
    * Check the expiry of the current plan, if it's future is resolved. If we are within 1s of
    * expiry, call startBackgroundRefresh with the version of the latest PrepareQuery.
    */
-  void backgroundRefreshIfNeeded() {
+  synchronized void backgroundRefreshIfNeeded() {
     PrepareQueryState localState = this.currentState.get();
     if (localState.maybeBackgroundRefresh().isPresent()) {
       // We already have an ongoing refresh
@@ -183,7 +183,7 @@ void backgroundRefreshIfNeeded() {
    * Returns the most recently refreshed PreparedQueryData. It may still be refreshing if the
    * previous plan has expired.
    */
-  public PreparedQueryData getLatestPrepareResponse() {
+  public synchronized PreparedQueryData getLatestPrepareResponse() {
     PrepareQueryState localState = currentState.get();
     if (localState.maybeBackgroundRefresh().isPresent()
         && localState.maybeBackgroundRefresh().get().prepareFuture().isDone()) {

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/exporter/BigtableCloudMonitoringExporter.java

@@ -79,7 +79,7 @@ public class BigtableCloudMonitoringExporter implements MetricExporter {
   private final MetricServiceClient client;
 
   private final AtomicReference<State> state;
-  private CompletableResultCode lastExportCode;
+  private AtomicReference<CompletableResultCode> lastExportCode = new AtomicReference<>();
   private final AtomicBoolean exportFailureLogged = new AtomicBoolean(false);
 
   private enum State {
@@ -149,8 +149,9 @@ public void close() {
   public CompletableResultCode export(Collection<MetricData> metricData) {
     Preconditions.checkState(state.get() != State.Closed, "Exporter is closed");
 
-    lastExportCode = doExport(metricData);
-    return lastExportCode;
+    CompletableResultCode result = doExport(metricData);
+    lastExportCode.set(result);
+    return result;
   }
 
   private CompletableResultCode doExport(Collection<MetricData> metricData) {
@@ -231,8 +232,8 @@ private List<ApiFuture<Empty>> exportTimeSeries(
 
   @Override
   public CompletableResultCode flush() {
-    if (lastExportCode != null) {
-      return lastExportCode;
+    if (lastExportCode.get() != null) {
+      return lastExportCode.get();
     }
     return CompletableResultCode.ofSuccess();
   }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/metrics/ClientDpCompatGuage.java

@@ -64,7 +64,7 @@ public void recordSuccess(ClientInfo clientInfo, String ipPreference) {
     public void recordFailure(
         ClientInfo clientInfo, DirectAccessInvestigator.FailureReason reason) {
       instrument.set(
-          1,
+          0,
           getSchema()
               .createResourceAttrs(clientInfo)
               .put(MetricLabels.DP_REASON_KEY, reason.getValue())

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImpl.java

@@ -507,11 +507,12 @@ private void onSessionClose(
       // vRPCs failing with consecutive failures should fail
       VRpcResult result =
           VRpcResult.createRejectedError(
-              Status.UNAVAILABLE.withDescription(
-                  String.format(
-                      "Session failed with consecutive failures. Most recent server status: %s,"
-                          + " metadata: %s.",
-                      status, trailers)));
+              Status.fromCode(status.getCode())
+                  .withDescription(
+                      String.format(
+                          "Session failed with consecutive failures. Most recent server status: %s,"
+                              + " metadata: %s.",
+                          status, trailers)));
       for (PendingVRpc<?, ?> vrpc : toBeClosed) {
         try {
           vrpc.getListener().onClose(result);

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/DynamicFlowControlStats.java

@@ -94,7 +94,7 @@ synchronized void update(long value) {
       weightedCount += weight;
     }
 
-    double getMean() {
+    synchronized double getMean() {
       return mean;
     }
 

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImplTest.java

@@ -206,7 +206,7 @@ public void consecutiveFailuresTest() {
     assertThat(exception)
         .hasMessageThat()
         .contains(
-            "UNAVAILABLE: Session failed with consecutive failures. Most recent server status:"
+            "INTERNAL: Session failed with consecutive failures. Most recent server status:"
                 + " Status{code=INTERNAL, description=fake internal error. PeerInfo:"
                 + " transport_type: TRANSPORT_TYPE_SESSION_UNKNOWN, cause=null}");
   }