[Dataflow Streaming] Remove rolled out Windmill isolated channels flag (#37844)

Author: arunpandianp

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java

@@ -126,7 +126,9 @@ public interface DataflowStreamingPipelineOptions extends PipelineOptions {
 
   void setWindmillMessagesBetweenIsReadyChecks(int value);
 
-  @Description("If true, a most a single active rpc will be used per channel.")
+  /** @deprecated since 2.73.0 */
+  @Deprecated
+  @Description("Unused flag.")
   Boolean getUseWindmillIsolatedChannels();
 
   void setUseWindmillIsolatedChannels(Boolean value);

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

@@ -725,7 +725,7 @@ public static StreamingDataflowWorker fromOptions(DataflowWorkerHarnessOptions o
           Function<ComputationConfig.Fetcher, ComputationStateCache> computationStateCacheFactory) {
     if (options.isEnableStreamingEngine()) {
       GrpcDispatcherClient dispatcherClient =
-          GrpcDispatcherClient.create(options, new WindmillStubFactoryFactoryImpl(options));
+          GrpcDispatcherClient.create(new WindmillStubFactoryFactoryImpl(options));
       ComputationConfig.Fetcher configFetcher =
           StreamingEngineComputationConfigFetcher.create(
               options.getGlobalConfigRefreshPeriod().getMillis(), dataflowServiceClient);
@@ -753,7 +753,7 @@ public static StreamingDataflowWorker fromOptions(DataflowWorkerHarnessOptions o
     if (options.getWindmillServiceEndpoint() != null
         || options.getLocalWindmillHostport().startsWith("grpc:")) {
       GrpcDispatcherClient dispatcherClient =
-          GrpcDispatcherClient.create(options, new WindmillStubFactoryFactoryImpl(options));
+          GrpcDispatcherClient.create(new WindmillStubFactoryFactoryImpl(options));
       GrpcWindmillStreamFactory windmillStreamFactory =
           windmillStreamFactoryBuilder
               .setHealthCheckIntervalMillis(
@@ -920,7 +920,7 @@ static StreamingDataflowWorker forTesting(
         createGrpcwindmillStreamFactoryBuilder(options, 1)
             .setProcessHeartbeatResponses(
                 new WorkHeartbeatResponseProcessor(computationStateCache::get));
-    GrpcDispatcherClient grpcDispatcherClient = GrpcDispatcherClient.create(options, stubFactory);
+    GrpcDispatcherClient grpcDispatcherClient = GrpcDispatcherClient.create(stubFactory);
     grpcDispatcherClient.consumeWindmillDispatcherEndpoints(
         ImmutableSet.<HostAndPort>builder()
             .add(HostAndPort.fromHost("StreamingDataflowWorkerTest"))

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcDispatcherClient.java

@@ -28,11 +28,9 @@
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
-import org.apache.beam.runners.dataflow.options.DataflowWorkerHarnessOptions;
 import org.apache.beam.runners.dataflow.worker.streaming.config.StreamingGlobalConfig;
 import org.apache.beam.runners.dataflow.worker.windmill.CloudWindmillMetadataServiceV1Alpha1Grpc;
 import org.apache.beam.runners.dataflow.worker.windmill.CloudWindmillMetadataServiceV1Alpha1Grpc.CloudWindmillMetadataServiceV1Alpha1Stub;
@@ -66,42 +64,25 @@ public class GrpcDispatcherClient {
   @GuardedBy("this")
   private final Random rand;
 
-  private final WindmillStubFactoryFactory windmillStubFactoryFactory;
-
-  private final AtomicReference<WindmillStubFactory> windmillStubFactory = new AtomicReference<>();
-
-  private final AtomicBoolean useIsolatedChannels = new AtomicBoolean();
-  private final boolean reactToIsolatedChannelsJobSetting;
+  private final WindmillStubFactory windmillStubFactory;
 
   private GrpcDispatcherClient(
-      DataflowWorkerHarnessOptions options,
       WindmillStubFactoryFactory windmillStubFactoryFactory,
       DispatcherStubs initialDispatcherStubs,
       Random rand) {
-    this.windmillStubFactoryFactory = windmillStubFactoryFactory;
-    if (options.getUseWindmillIsolatedChannels() != null) {
-      this.useIsolatedChannels.set(options.getUseWindmillIsolatedChannels());
-      this.reactToIsolatedChannelsJobSetting = false;
-    } else {
-      this.useIsolatedChannels.set(false);
-      this.reactToIsolatedChannelsJobSetting = true;
-    }
-    this.windmillStubFactory.set(
-        windmillStubFactoryFactory.makeWindmillStubFactory(useIsolatedChannels.get()));
+    this.windmillStubFactory = windmillStubFactoryFactory.makeWindmillStubFactory();
     this.rand = rand;
     this.dispatcherStubs = new AtomicReference<>(initialDispatcherStubs);
     this.onInitializedEndpoints = new CountDownLatch(1);
   }
 
-  public static GrpcDispatcherClient create(
-      DataflowWorkerHarnessOptions options, WindmillStubFactoryFactory windmillStubFactoryFactory) {
+  public static GrpcDispatcherClient create(WindmillStubFactoryFactory windmillStubFactoryFactory) {
     return new GrpcDispatcherClient(
-        options, windmillStubFactoryFactory, DispatcherStubs.empty(), new Random());
+        windmillStubFactoryFactory, DispatcherStubs.empty(), new Random());
   }
 
   @VisibleForTesting
   public static GrpcDispatcherClient forTesting(
-      DataflowWorkerHarnessOptions options,
       WindmillStubFactoryFactory windmillStubFactoryFactory,
       List<CloudWindmillServiceV1Alpha1Stub> windmillServiceStubs,
       List<CloudWindmillMetadataServiceV1Alpha1Stub> windmillMetadataServiceStubs,
@@ -110,7 +91,6 @@ public static GrpcDispatcherClient forTesting(
         dispatcherEndpoints.size() == windmillServiceStubs.size()
             && windmillServiceStubs.size() == windmillMetadataServiceStubs.size());
     return new GrpcDispatcherClient(
-        options,
         windmillStubFactoryFactory,
         DispatcherStubs.create(
             dispatcherEndpoints, windmillServiceStubs, windmillMetadataServiceStubs),
@@ -172,31 +152,17 @@ public void onJobConfig(StreamingGlobalConfig config) {
       LOG.warn("Dispatcher client received empty windmill service endpoints from global config");
       return;
     }
-    boolean forceRecreateStubs = false;
-    if (reactToIsolatedChannelsJobSetting) {
-      boolean useIsolatedChannels = config.userWorkerJobSettings().getUseWindmillIsolatedChannels();
-      if (this.useIsolatedChannels.getAndSet(useIsolatedChannels) != useIsolatedChannels) {
-        windmillStubFactory.set(
-            windmillStubFactoryFactory.makeWindmillStubFactory(useIsolatedChannels));
-        forceRecreateStubs = true;
-      }
-    }
-    consumeWindmillDispatcherEndpoints(config.windmillServiceEndpoints(), forceRecreateStubs);
+    consumeWindmillDispatcherEndpoints(config.windmillServiceEndpoints());
   }
 
   public synchronized void consumeWindmillDispatcherEndpoints(
       ImmutableSet<HostAndPort> dispatcherEndpoints) {
-    consumeWindmillDispatcherEndpoints(dispatcherEndpoints, /* forceRecreateStubs= */ false);
-  }
-
-  private synchronized void consumeWindmillDispatcherEndpoints(
-      ImmutableSet<HostAndPort> dispatcherEndpoints, boolean forceRecreateStubs) {
     ImmutableSet<HostAndPort> currentDispatcherEndpoints =
         dispatcherStubs.get().dispatcherEndpoints();
     Preconditions.checkArgument(
         dispatcherEndpoints != null && !dispatcherEndpoints.isEmpty(),
         "Cannot set dispatcher endpoints to nothing.");
-    if (!forceRecreateStubs && currentDispatcherEndpoints.equals(dispatcherEndpoints)) {
+    if (currentDispatcherEndpoints.equals(dispatcherEndpoints)) {
       // The endpoints are equal don't recreate the stubs.
       return;
     }
@@ -207,7 +173,7 @@ private synchronized void consumeWindmillDispatcherEndpoints(
     }
 
     LOG.info("Initializing Streaming Engine GRPC client for endpoints: {}", dispatcherEndpoints);
-    dispatcherStubs.set(DispatcherStubs.create(dispatcherEndpoints, windmillStubFactory.get()));
+    dispatcherStubs.set(DispatcherStubs.create(dispatcherEndpoints, windmillStubFactory));
     onInitializedEndpoints.countDown();
   }
 

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcWindmillServer.java

@@ -166,7 +166,6 @@ static GrpcWindmillServer newTestInstance(
     Set<HostAndPort> dispatcherEndpoints = Sets.newHashSet(HostAndPort.fromHost(name));
     GrpcDispatcherClient dispatcherClient =
         GrpcDispatcherClient.forTesting(
-            testOptions,
             windmillStubFactoryFactory,
             windmillServiceStubs,
             windmillMetadataServiceStubs,
@@ -198,7 +197,7 @@ static GrpcWindmillServer newApplianceTestInstance(
             options,
             GrpcWindmillStreamFactory.of(createJobHeader(options, 1)).build(),
             // No-op, Appliance does not use Dispatcher to call Streaming Engine.
-            GrpcDispatcherClient.create(options, windmillStubFactoryFactory));
+            GrpcDispatcherClient.create(windmillStubFactoryFactory));
     testServer.syncApplianceStub = createWindmillApplianceStubWithDeadlineInterceptor(channel);
     return testServer;
   }

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/WindmillStubFactoryFactory.java

@@ -21,5 +21,5 @@
 
 @Internal
 public interface WindmillStubFactoryFactory {
-  WindmillStubFactory makeWindmillStubFactory(boolean useIsolatedChannels);
+  WindmillStubFactory makeWindmillStubFactory();
 }

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/WindmillStubFactoryFactoryImpl.java

@@ -33,24 +33,19 @@ public WindmillStubFactoryFactoryImpl(DataflowWorkerHarnessOptions workerOptions
   }
 
   @Override
-  public WindmillStubFactory makeWindmillStubFactory(boolean useIsolatedChannels) {
+  public WindmillStubFactory makeWindmillStubFactory() {
     ChannelCache channelCache =
         ChannelCache.create(
             (flowControlSettings, serviceAddress) ->
                 // IsolationChannel will create and manage separate RPC channels to the same
                 // serviceAddress via calling the channelFactory, else just directly return the
                 // RPC channel.
-                useIsolatedChannels
-                    ? IsolationChannel.create(
-                        () ->
-                            remoteChannel(
-                                serviceAddress.getServiceAddress(),
-                                windmillServiceRpcChannelAliveTimeoutSec,
-                                flowControlSettings))
-                    : remoteChannel(
-                        serviceAddress.getServiceAddress(),
-                        windmillServiceRpcChannelAliveTimeoutSec,
-                        flowControlSettings));
+                IsolationChannel.create(
+                    () ->
+                        remoteChannel(
+                            serviceAddress.getServiceAddress(),
+                            windmillServiceRpcChannelAliveTimeoutSec,
+                            flowControlSettings)));
     return ChannelCachingRemoteStubFactory.create(gcpCredential, channelCache);
   }
 }

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarnessTest.java

@@ -36,7 +36,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
-import org.apache.beam.runners.dataflow.options.DataflowWorkerHarnessOptions;
 import org.apache.beam.runners.dataflow.worker.util.MemoryMonitor;
 import org.apache.beam.runners.dataflow.worker.windmill.CloudWindmillMetadataServiceV1Alpha1Grpc;
 import org.apache.beam.runners.dataflow.worker.windmill.CloudWindmillServiceV1Alpha1Grpc;
@@ -58,7 +57,6 @@
 import org.apache.beam.runners.dataflow.worker.windmill.work.budget.GetWorkBudget;
 import org.apache.beam.runners.dataflow.worker.windmill.work.budget.GetWorkBudgetDistributor;
 import org.apache.beam.runners.dataflow.worker.windmill.work.budget.GetWorkBudgetSpender;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.Server;
 import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.inprocess.InProcessServerBuilder;
 import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.inprocess.InProcessSocketAddress;
@@ -112,7 +110,6 @@ public class FanOutStreamingEngineWorkerHarnessTest {
           () -> grpcCleanup.register(WindmillChannels.inProcessChannel(CHANNEL_NAME)));
   private final GrpcDispatcherClient dispatcherClient =
       GrpcDispatcherClient.forTesting(
-          PipelineOptionsFactory.as(DataflowWorkerHarnessOptions.class),
           new FakeWindmillStubFactoryFactory(stubFactory),
           new ArrayList<>(),
           new ArrayList<>(),

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcDispatcherClientTest.java

@@ -29,7 +29,7 @@ public FakeWindmillStubFactoryFactory(WindmillStubFactory windmillStubFactory) {
   }
 
   @Override
-  public WindmillStubFactory makeWindmillStubFactory(boolean useIsolatedChannels) {
+  public WindmillStubFactory makeWindmillStubFactory() {
     return windmillStubFactory;
   }
 }

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/testing/FakeWindmillStubFactoryFactory.java

@@ -966,16 +966,15 @@ enum ConnectivityType {
 
 // Settings to control runtime behavior of the java runner v1 user worker.
 message UserWorkerRunnerV1Settings {
-  // If true, use separate channels for each windmill RPC.
-  optional bool use_windmill_isolated_channels = 1 [default = true];
-
   // If true, use separate streaming RPC for windmill heartbeats and state reads.
   optional bool use_separate_windmill_heartbeat_streams = 2 [default = true];
 
   optional UserWorkerGrpcFlowControlSettings flow_control_settings = 3;
 
   optional ConnectivityType connectivity_type = 4
     [default = CONNECTIVITY_TYPE_DEFAULT];
+
+  reserved 1;
 }
 
 service WindmillAppliance {