From f2700d66157b9d3212b2311bf67bfba8040248b6 Mon Sep 17 00:00:00 2001
From: Sania Parveen <saniap@google.com>
Date: Wed, 5 Aug 2026 21:25:45 +0000
Subject: [PATCH 1/4] Make primary channel timeout configurable

---
 .../worker/StreamingDataflowWorker.java       | 14 ++-
 .../client/grpc/stubs/FailoverChannel.java    | 51 +++++++----
 .../grpc/stubs/FailoverChannelTest.java       | 88 ++++++++++++++++++-
 .../windmill/src/main/proto/windmill.proto    |  4 +
 4 files changed, 137 insertions(+), 20 deletions(-)

diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
index 9e82343474c6..b9217b520749 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
@@ -34,6 +34,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -825,6 +826,7 @@ private static ChannelCache createChannelCache(
       DataflowWorkerHarnessOptions workerOptions,
       ComputationConfig.Fetcher configFetcher,
       GrpcDispatcherClient dispatcherClient) {
+    AtomicLong primaryNotReadyWaitNanos = new AtomicLong(TimeUnit.SECONDS.toNanos(15));
     ChannelCache channelCache =
         ChannelCache.create(
             (currentFlowControlSettings, serviceAddress) -> {
@@ -844,16 +846,20 @@ private static ChannelCache createChannelCache(
                                   workerOptions.getWindmillServiceRpcChannelAliveTimeoutSec(),
                                   currentFlowControlSettings),
                           MoreCallCredentials.from(
-                              new VendoredCredentialsAdapter(workerOptions.getGcpCredential()))),
+                              new VendoredCredentialsAdapter(workerOptions.getGcpCredential())),
+                          primaryNotReadyWaitNanos::get),
                   currentFlowControlSettings.getOnReadyThresholdBytes());
             });
 
     configFetcher
         .getGlobalConfigHandle()
         .registerConfigObserver(
-            config ->
-                channelCache.consumeFlowControlSettings(
-                    config.userWorkerJobSettings().getFlowControlSettings()));
+            config -> {
+              primaryNotReadyWaitNanos.set(
+                  config.userWorkerJobSettings().getPrimaryNotReadyWaitNanos());
+              channelCache.consumeFlowControlSettings(
+                  config.userWorkerJobSettings().getFlowControlSettings());
+            });
     return channelCache;
   }
 
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java
index faa08c497c8f..8bb83e12eaf5 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
 
+import java.net.http.WebSocket.Listener;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -45,9 +46,9 @@
  * <p>Routes requests to either primary or fallback channel based on two independent failover modes:
  *
  * <ul>
- *   <li><b>Connection Status Failover:</b> If the primary channel is not ready for 10+ seconds
- *       (e.g., during network issues), routes to fallback channel. Switches back as soon as the
- *       primary channel becomes READY again.
+ *   <li><b>Connection Status Failover:</b> If the primary channel is not ready for the configured
+ *       wait time (e.g., during network issues), routes to fallback channel. Switches back as soon
+ *       as the primary channel becomes READY again.
  *   <li><b>RPC Failover:</b> If primary channel RPCs fail continuously with transient errors
  *       ({@link Status.Code#UNAVAILABLE} or {@link Status.Code#UNKNOWN}), or with {@link
  *       Status.Code#DEADLINE_EXCEEDED} before receiving any response (indicating the connection was
@@ -61,7 +62,6 @@ public final class FailoverChannel extends ManagedChannel {
   private static final AtomicInteger CHANNEL_ID_COUNTER = new AtomicInteger(0);
   // Time to wait before retrying the primary channel after an RPC-based fallback.
   private static final long FALLBACK_COOLING_PERIOD_NANOS = TimeUnit.HOURS.toNanos(1);
-  private static final long PRIMARY_NOT_READY_WAIT_NANOS = TimeUnit.SECONDS.toNanos(10);
   // Minimum duration of continuous RPC failures required before switching to fallback.
   private static final long RPC_FAILURE_THRESHOLD_NANOS = TimeUnit.SECONDS.toNanos(30);
 
@@ -96,10 +96,15 @@ private static final class FailoverState {
 
     private final int channelId;
     private final long rpcFailureThresholdNanos;
+    private final LongSupplier primaryNotReadyWaitNanosSupplier;
 
-    FailoverState(int channelId, long rpcFailureThresholdNanos) {
+    FailoverState(
+        int channelId,
+        long rpcFailureThresholdNanos,
+        LongSupplier primaryNotReadyWaitNanosSupplier) {
       this.channelId = channelId;
       this.rpcFailureThresholdNanos = rpcFailureThresholdNanos;
+      this.primaryNotReadyWaitNanosSupplier = primaryNotReadyWaitNanosSupplier;
     }
 
     /**
@@ -121,7 +126,7 @@ synchronized boolean computeUseFallback(long nowNanos) {
       if (!useFallbackDueToRPC
           && !useFallbackDueToState
           && primaryNotReadySinceNanos >= 0
-          && nowNanos - primaryNotReadySinceNanos > PRIMARY_NOT_READY_WAIT_NANOS) {
+          && nowNanos - primaryNotReadySinceNanos > primaryNotReadyWaitNanosSupplier.getAsLong()) {
         useFallbackDueToState = true;
         LOG.warn(
             "[channel-{}] Primary connection unavailable. Switching to secondary connection.",
@@ -193,11 +198,13 @@ private FailoverChannel(
       Supplier<ManagedChannel> fallbackSupplier,
       @Nullable CallCredentials fallbackCallCredentials,
       LongSupplier nanoClock,
-      long rpcFailureThresholdNanos) {
+      long rpcFailureThresholdNanos,
+      LongSupplier primaryNotReadyWaitNanosSupplier) {
     this.primary = primary;
     this.fallbackSupplier = Suppliers.memoize(fallbackSupplier::get);
     this.channelId = CHANNEL_ID_COUNTER.getAndIncrement();
-    this.state = new FailoverState(channelId, rpcFailureThresholdNanos);
+    this.state =
+        new FailoverState(channelId, rpcFailureThresholdNanos, primaryNotReadyWaitNanosSupplier);
     this.fallbackCallCredentials = fallbackCallCredentials;
     this.nanoClock = nanoClock;
     // Register callback to monitor primary channel state changes
@@ -207,13 +214,15 @@ private FailoverChannel(
   public static FailoverChannel create(
       ManagedChannel primary,
       Supplier<ManagedChannel> fallbackSupplier,
-      CallCredentials fallbackCallCredentials) {
+      CallCredentials fallbackCallCredentials,
+      LongSupplier primaryNotReadyWaitNanosSupplier) {
     return new FailoverChannel(
         primary,
         fallbackSupplier,
         fallbackCallCredentials,
         System::nanoTime,
-        RPC_FAILURE_THRESHOLD_NANOS);
+        RPC_FAILURE_THRESHOLD_NANOS,
+        primaryNotReadyWaitNanosSupplier);
   }
 
   static FailoverChannel forTest(
@@ -221,9 +230,15 @@ static FailoverChannel forTest(
       ManagedChannel fallback,
       CallCredentials fallbackCallCredentials,
       LongSupplier nanoClock,
-      long rpcFailureThresholdNanos) {
+      long rpcFailureThresholdNanos,
+      LongSupplier primaryNotReadyWaitNanosSupplier) {
     return new FailoverChannel(
-        primary, () -> fallback, fallbackCallCredentials, nanoClock, rpcFailureThresholdNanos);
+        primary,
+        () -> fallback,
+        fallbackCallCredentials,
+        nanoClock,
+        rpcFailureThresholdNanos,
+        primaryNotReadyWaitNanosSupplier);
   }
 
   /** Returns the fallback channel, creating it from the supplier at most once. */
@@ -399,7 +414,12 @@ private void registerPrimaryStateChangeListener() {
         // never transitions, markPrimaryNotReady() would never be called and state-based
         // failover would not trigger even after the grace period.
         if (currentState == ConnectivityState.READY || currentState == ConnectivityState.IDLE) {
-          state.markPrimaryReady();
+          if (state.markPrimaryReady()) {
+            LOG.info(
+                "[channel-{}] Primary channel observed healthy during state change registration;"
+                    + " switching back from fallback.",
+                channelId);
+          }
         } else {
           // Seed the not-ready timer even if there is no future state transition.
           state.markPrimaryNotReady(nanoClock.getAsLong());
@@ -426,11 +446,12 @@ private void onPrimaryStateChanged() {
     if (newState == ConnectivityState.READY || newState == ConnectivityState.IDLE) {
       if (state.markPrimaryReady()) {
         LOG.info(
-            "[channel-{}] Primary channel recovered; switching back from fallback.", channelId);
+            "[channel-{}] Primary channel observed healthy during state change registration; switching back from fallback.",
+            channelId);
       }
     } else {
       // Primary is not ready; start the grace period timer so computeUseFallback can
-      // switch to fallback once PRIMARY_NOT_READY_WAIT_NANOS elapses.
+      // switch to fallback once the configured wait time elapses.
       state.markPrimaryNotReady(nanoClock.getAsLong());
     }
 
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannelTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannelTest.java
index 9a46e5bc5489..dfbbf9a695aa 100644
--- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannelTest.java
+++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannelTest.java
@@ -87,7 +87,8 @@ private static FailoverChannel createForTest(
         fallback,
         fallbackCallCredentials,
         nanoClock,
-        rpcFailureThresholdNanos != null ? rpcFailureThresholdNanos : 0L);
+        rpcFailureThresholdNanos != null ? rpcFailureThresholdNanos : 0L,
+        () -> TimeUnit.SECONDS.toNanos(10));
   }
 
   /**
@@ -290,6 +291,91 @@ public void testStateFallbackAfterPrimaryNotReady() {
     verify(mockFallbackChannel).newCall(any(), any());
   }
 
+  @Test
+  public void testTimeoutThresholdDecreaseTriggersFallbackEarlier() {
+    ManagedChannel mockChannel = mock(ManagedChannel.class);
+    ManagedChannel mockFallbackChannel = mock(ManagedChannel.class);
+    when(mockChannel.newCall(any(), any())).thenReturn(mock(ClientCall.class));
+    when(mockFallbackChannel.newCall(any(), any())).thenReturn(mock(ClientCall.class));
+    // Simulate primary being TRANSIENT_FAILURE from the start.
+    when(mockChannel.getState(false)).thenReturn(ConnectivityState.TRANSIENT_FAILURE);
+
+    AtomicLong time = new AtomicLong(0);
+    // Start with 10s timeout
+    AtomicLong timeoutThreshold = new AtomicLong(TimeUnit.SECONDS.toNanos(10));
+
+    // Constructor seeds timer at time=0.
+    FailoverChannel failoverChannel =
+        FailoverChannel.forTest(
+            mockChannel, mockFallbackChannel, null, time::get, 0L, timeoutThreshold::get);
+
+    // Call at time=0. elapsed 0 <= 10s -> false.
+    failoverChannel.newCall(methodDescriptor, CallOptions.DEFAULT);
+    verify(mockChannel).newCall(any(), any());
+
+    // Advance time by 5 seconds.
+    time.addAndGet(TimeUnit.SECONDS.toNanos(5));
+
+    // Call at time=5s. elapsed 5s <= 10s -> false.
+    // Primary is used.
+    failoverChannel.newCall(methodDescriptor, CallOptions.DEFAULT);
+    verify(mockChannel, org.mockito.Mockito.times(2)).newCall(any(), any());
+
+    // Decrease threshold to 2 seconds.
+    timeoutThreshold.set(TimeUnit.SECONDS.toNanos(2));
+
+    // Call at time=5s. elapsed 5s > 2s -> true.
+    // Fallback is used.
+    failoverChannel.newCall(methodDescriptor, CallOptions.DEFAULT);
+    verify(mockFallbackChannel).newCall(any(), any());
+  }
+
+  @Test
+  public void testTimeoutThresholdIncreaseDelaysFallback() {
+    ManagedChannel mockChannel = mock(ManagedChannel.class);
+    ManagedChannel mockFallbackChannel = mock(ManagedChannel.class);
+    when(mockChannel.newCall(any(), any())).thenReturn(mock(ClientCall.class));
+    when(mockFallbackChannel.newCall(any(), any())).thenReturn(mock(ClientCall.class));
+    // Simulate primary being TRANSIENT_FAILURE from the start.
+    when(mockChannel.getState(false)).thenReturn(ConnectivityState.TRANSIENT_FAILURE);
+
+    AtomicLong time = new AtomicLong(0);
+    // Start with 10s timeout
+    AtomicLong timeoutThreshold = new AtomicLong(TimeUnit.SECONDS.toNanos(10));
+
+    // Constructor seeds timer at time=0.
+    FailoverChannel failoverChannel =
+        FailoverChannel.forTest(
+            mockChannel, mockFallbackChannel, null, time::get, 0L, timeoutThreshold::get);
+
+    // Advance time by 9 seconds.
+    time.addAndGet(TimeUnit.SECONDS.toNanos(9));
+
+    // Call at time=9s. elapsed 9s <= 10s -> false.
+    // Primary is used.
+    failoverChannel.newCall(methodDescriptor, CallOptions.DEFAULT);
+    verify(mockChannel).newCall(any(), any());
+
+    // Increase threshold to 20 seconds.
+    timeoutThreshold.set(TimeUnit.SECONDS.toNanos(20));
+
+    // Advance time by 5 seconds (total 14s).
+    time.addAndGet(TimeUnit.SECONDS.toNanos(5));
+
+    // Call at time=14s. elapsed 14s <= 20s -> false.
+    // Still routes to primary because of increased threshold
+    failoverChannel.newCall(methodDescriptor, CallOptions.DEFAULT);
+    verify(mockChannel, org.mockito.Mockito.times(2)).newCall(any(), any());
+
+    // Advance time by 7s (total 21s).
+    time.addAndGet(TimeUnit.SECONDS.toNanos(7));
+
+    // Call at time=21s. elapsed 21s > 20s -> true.
+    // Fallback is used.
+    failoverChannel.newCall(methodDescriptor, CallOptions.DEFAULT);
+    verify(mockFallbackChannel).newCall(any(), any());
+  }
+
   @Test
   public void testStateFallbackWhenPrimaryStartsNonReadyWithoutTransition() {
     // Primary starts in a non-ready state and stays there. Even without a state transition,
diff --git a/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto b/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto
index aaa09c105fc3..fcef6f0fea17 100644
--- a/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto
+++ b/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto
@@ -995,6 +995,10 @@ message UserWorkerRunnerV1Settings {
 
   optional int64 max_cached_entry_bytes = 5 [default = -1];
 
+  // Time to wait before switching to fallback connectivity if primary is not ready.
+  // Only used if direcpath is enabled for the job. Default is 15 seconds.
+  optional int64 primary_not_ready_wait_nanos = 6 [default = 15000000000];
+
   reserved 1, 2;
 }
 

From 3becc915c263e4fce5f9ea1b3b957cb4be10d6f5 Mon Sep 17 00:00:00 2001
From: Sania Parveen <saniap@google.com>
Date: Wed, 5 Aug 2026 22:23:54 +0000
Subject: [PATCH 2/4] Update setting name primary_not_ready_wait_nanos to
 directpath_primary_not_ready_wait_nanos

---
 .../runners/dataflow/worker/StreamingDataflowWorker.java  | 8 ++++----
 .../worker/windmill/src/main/proto/windmill.proto         | 2 +-
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
index 72a626527309..32603d862c5c 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
@@ -829,7 +829,7 @@ private static ChannelCache createChannelCache(
       DataflowWorkerHarnessOptions workerOptions,
       ComputationConfig.Fetcher configFetcher,
       GrpcDispatcherClient dispatcherClient) {
-    AtomicLong primaryNotReadyWaitNanos = new AtomicLong(TimeUnit.SECONDS.toNanos(15));
+    AtomicLong directpathPrimaryNotReadyWaitNanos = new AtomicLong(TimeUnit.SECONDS.toNanos(15));
     ChannelCache channelCache =
         ChannelCache.create(
             (currentFlowControlSettings, serviceAddress) -> {
@@ -850,7 +850,7 @@ private static ChannelCache createChannelCache(
                                   currentFlowControlSettings),
                           MoreCallCredentials.from(
                               new VendoredCredentialsAdapter(workerOptions.getGcpCredential())),
-                          primaryNotReadyWaitNanos::get),
+                          directpathPrimaryNotReadyWaitNanos::get),
                   currentFlowControlSettings.getOnReadyThresholdBytes());
             });
 
@@ -858,8 +858,8 @@ private static ChannelCache createChannelCache(
         .getGlobalConfigHandle()
         .registerConfigObserver(
             config -> {
-              primaryNotReadyWaitNanos.set(
-                  config.userWorkerJobSettings().getPrimaryNotReadyWaitNanos());
+              directpathPrimaryNotReadyWaitNanos.set(
+                  config.userWorkerJobSettings().getDirectpathPrimaryNotReadyWaitNanos());
               channelCache.consumeFlowControlSettings(
                   config.userWorkerJobSettings().getFlowControlSettings());
             });
diff --git a/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto b/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto
index 4f2cfce4686a..3bb644d5199a 100644
--- a/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto
+++ b/runners/google-cloud-dataflow-java/worker/windmill/src/main/proto/windmill.proto
@@ -1020,7 +1020,7 @@ message UserWorkerRunnerV1Settings {
 
   // Time to wait before switching to fallback connectivity if primary is not ready.
   // Only used if direcpath is enabled for the job. Default is 15 seconds.
-  optional int64 primary_not_ready_wait_nanos = 6 [default = 15000000000];
+  optional int64 directpath_primary_not_ready_wait_nanos = 6 [default = 15000000000];
 
   reserved 1, 2;
 }

From cece286110cb2f3be47a13d0654ce4b820bb138f Mon Sep 17 00:00:00 2001
From: Sania Parveen <saniap@google.com>
Date: Wed, 5 Aug 2026 22:47:03 +0000
Subject: [PATCH 3/4] Addresing review comments

---
 .../runners/dataflow/worker/StreamingDataflowWorker.java   | 7 ++++---
 .../worker/windmill/client/grpc/stubs/FailoverChannel.java | 1 -
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
index 32603d862c5c..1f2452b68cd2 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
@@ -173,6 +173,8 @@ public final class StreamingDataflowWorker {
   private static final String CHANNELZ_PATH = "/channelz";
   private static final String BEAM_FN_API_EXPERIMENT = "beam_fn_api";
   private static final String ELEMENT_METADATA_SUPPORTED_EXPERIMENT = "element_metadata_supported";
+  private static final AtomicLong DIRECTPATH_PRIMARY_NOT_READY_WAIT_NANOS =
+      new AtomicLong(TimeUnit.SECONDS.toNanos(15));
 
   @SuppressWarnings("unused")
   private static final String STREAMING_ENGINE_USE_JOB_SETTINGS_FOR_HEARTBEAT_POOL_EXPERIMENT =
@@ -829,7 +831,6 @@ private static ChannelCache createChannelCache(
       DataflowWorkerHarnessOptions workerOptions,
       ComputationConfig.Fetcher configFetcher,
       GrpcDispatcherClient dispatcherClient) {
-    AtomicLong directpathPrimaryNotReadyWaitNanos = new AtomicLong(TimeUnit.SECONDS.toNanos(15));
     ChannelCache channelCache =
         ChannelCache.create(
             (currentFlowControlSettings, serviceAddress) -> {
@@ -850,7 +851,7 @@ private static ChannelCache createChannelCache(
                                   currentFlowControlSettings),
                           MoreCallCredentials.from(
                               new VendoredCredentialsAdapter(workerOptions.getGcpCredential())),
-                          directpathPrimaryNotReadyWaitNanos::get),
+                          DIRECTPATH_PRIMARY_NOT_READY_WAIT_NANOS::get),
                   currentFlowControlSettings.getOnReadyThresholdBytes());
             });
 
@@ -858,7 +859,7 @@ private static ChannelCache createChannelCache(
         .getGlobalConfigHandle()
         .registerConfigObserver(
             config -> {
-              directpathPrimaryNotReadyWaitNanos.set(
+              DIRECTPATH_PRIMARY_NOT_READY_WAIT_NANOS.set(
                   config.userWorkerJobSettings().getDirectpathPrimaryNotReadyWaitNanos());
               channelCache.consumeFlowControlSettings(
                   config.userWorkerJobSettings().getFlowControlSettings());
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java
index 8bb83e12eaf5..d243ee5430f0 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/FailoverChannel.java
@@ -17,7 +17,6 @@
  */
 package org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs;
 
-import java.net.http.WebSocket.Listener;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;

From 046502e5ebc4f3705ce4c2818da9b5a7f3114bfe Mon Sep 17 00:00:00 2001
From: Sania Parveen <saniap@google.com>
Date: Wed, 5 Aug 2026 23:46:48 +0000
Subject: [PATCH 4/4] Trigger CI