Fix thread-safety in DelayedClientTransport and remove Thread.sleep in E2E delay tests

Author: AgraVator

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

@@ -401,7 +401,9 @@ private class PendingStream extends DelayedStream {
     private final Context context = Context.current();
     private final ClientStreamTracer[] tracers;
     private volatile Status lastPickStatus;
+    @GuardedBy("this")
     @Nullable private String activeDelayType;
+    @GuardedBy("this")
     @Nullable private String activeDelayReason;
 
     private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers,
@@ -425,7 +427,7 @@ private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers,
      * spans are ended and a new segment is initiated. If only {@code newReason} changes, a
      * structured transition event is appended to the active span without span re-creation.
      */
-    void updateDelay(@Nullable String newType, @Nullable String newReason) {
+    synchronized void updateDelay(@Nullable String newType, @Nullable String newReason) {
       if (!Objects.equals(activeDelayType, newType)) {
         // Delay type changed (e.g., from RLS lookup to connecting). End the previous delay.
         if (activeDelayType != null) {
@@ -453,7 +455,7 @@ void updateDelay(@Nullable String newType, @Nullable String newReason) {
     /**
      * Ends active attempt delay segment telemetry upon stream creation or stream cancellation.
      */
-    void endDelay() {
+    synchronized void endDelay() {
       if (activeDelayType != null) {
         for (ClientStreamTracer tracer : tracers) {
           tracer.recordAttemptDelayEnd();

opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryMetricsModuleTest.java

@@ -1690,11 +1690,11 @@ public Status acceptResolvedAddresses(LoadBalancer.ResolvedAddresses resolvedAdd
             helper.updateBalancingState(ConnectivityState.CONNECTING, new SubchannelPicker() {
               @Override
               public PickResult pickSubchannel(PickSubchannelArgs args) {
+                latch.countDown();
                 return PickResult.withNoResult("connecting",
                     "Simulated slow TLS handshake with backend");
               }
             });
-            latch.countDown();
             return Status.OK;
           }
 
@@ -1769,7 +1769,6 @@ public void shutdown() {}
       call.request(1);
 
       latch.await(5, TimeUnit.SECONDS);
-      Thread.sleep(50);
       call.cancel("End test delay segment", null);
     } finally {
       channel.shutdownNow();

opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java

@@ -481,11 +481,11 @@ public Status acceptResolvedAddresses(LoadBalancer.ResolvedAddresses resolvedAdd
             helper.updateBalancingState(ConnectivityState.CONNECTING, new SubchannelPicker() {
               @Override
               public PickResult pickSubchannel(PickSubchannelArgs args) {
+                latch.countDown();
                 return PickResult.withNoResult("connecting",
                     "Simulated slow TLS handshake with backend");
               }
             });
-            latch.countDown();
             return Status.OK;
           }
 
@@ -559,7 +559,6 @@ public void shutdown() {}
       call.request(1);
 
       latch.await(5, TimeUnit.SECONDS);
-      Thread.sleep(50);
       call.cancel("End test delay segment", null);
     } finally {
       channel.shutdownNow();