Merge branch 'apache:master' into fix-python-postcommit-30513

Author: aIbrahiim

.github/workflows/build_wheels.yml

@@ -229,15 +229,6 @@ jobs:
         ]
         # Keep in sync (remove asterisks) with PY_VERSIONS_FULL env var above - if changed, change that as well.
         py_version: ["cp310-", "cp311-", "cp312-", "cp313-"]
-        # The following exclude/include skips intermediate Python versions on GitHub hosted runners for PR run
-        is_pr:
-          - ${{ github.event_name == 'pull_request' }}
-        exclude:
-          - is_pr: true
-        include:
-          - os_python.os: "ubuntu-20.04"
-          - py_version: "cp310-"
-          - py_version: "cp313-"
     steps:
     - name: Download python source distribution from artifacts
       uses: actions/download-artifact@v5

CONTRIBUTING.md

@@ -41,6 +41,11 @@ These steps and instructions on getting started are outlined below as well.
 - A [GitHub](https://github.com/) account.
 - A Linux, macOS, or Microsoft Windows development environment.
 - Java JDK 11 (preferred, or 8, 17, 21) installed.
+  - General note:
+    - Set `JAVA_HOME` to the JDK **installation directory**, not the `bin` directory
+    - (Window for example: `C:\Program Files\Eclipse Adoptium\jdk-11.x.x`)
+    - If multiple JDK versions are installed, ensure Java 11 appears first on `PATH`
+    - The initial Gradle build may take 10–15 minutes due to dependency downloads
 - Latest [Go](https://golang.org) 1.x installed.
 - [Docker](https://www.docker.com/) installed for some tasks including building worker containers and testing changes to this website locally.
 - For SDK Development:

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

@@ -555,6 +555,11 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
       return timestamp();
     }
 
+    @Override
+    public CausedByDrain causedByDrain(DoFn<InputT, OutputT> doFn) {
+      return elem.causedByDrain();
+    }
+
     @Override
     public String timerId(DoFn<InputT, OutputT> doFn) {
       throw new UnsupportedOperationException(
@@ -831,6 +836,11 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
       return timestamp();
     }
 
+    @Override
+    public CausedByDrain causedByDrain(DoFn<InputT, OutputT> doFn) {
+      return causedByDrain;
+    }
+
     @Override
     public String timerId(DoFn<InputT, OutputT> doFn) {
       return timerId;
@@ -1119,6 +1129,11 @@ public Instant timestamp(DoFn<InputT, OutputT> doFn) {
       return timestamp;
     }
 
+    @Override
+    public CausedByDrain causedByDrain(DoFn<InputT, OutputT> doFn) {
+      throw new UnsupportedOperationException("CausedByDrain parameters are not supported.");
+    }
+
     @Override
     public String timerId(DoFn<InputT, OutputT> doFn) {
       throw new UnsupportedOperationException("Timer parameters are not supported.");

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

@@ -126,13 +126,16 @@ 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);
 
-  @Description(
-      "If true, separate streaming rpcs will be used for heartbeats instead of sharing streams with state reads.")
+  /** @deprecated since beam 2.73.0 */
+  @Deprecated
+  @Description("Unused Flag")
   Boolean getUseSeparateWindmillHeartbeatStreams();
 
   void setUseSeparateWindmillHeartbeatStreams(Boolean value);

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

@@ -54,7 +54,6 @@
 import org.apache.beam.runners.dataflow.worker.streaming.config.StreamingApplianceComputationConfigFetcher;
 import org.apache.beam.runners.dataflow.worker.streaming.config.StreamingEngineComputationConfigFetcher;
 import org.apache.beam.runners.dataflow.worker.streaming.config.StreamingGlobalConfig;
-import org.apache.beam.runners.dataflow.worker.streaming.config.StreamingGlobalConfigHandle;
 import org.apache.beam.runners.dataflow.worker.streaming.config.StreamingGlobalConfigHandleImpl;
 import org.apache.beam.runners.dataflow.worker.streaming.harness.FanOutStreamingEngineWorkerHarness;
 import org.apache.beam.runners.dataflow.worker.streaming.harness.SingleSourceWorkerHarness;
@@ -460,12 +459,7 @@ private StreamingWorkerHarnessFactoryOutput createSingleSourceWorkerHarness(
             windmillServer::getDataStream);
     GetDataClient getDataClient =
         new StreamPoolGetDataClient(getDataMetricTracker, getDataStreamPool);
-    HeartbeatSender heartbeatSender =
-        createStreamingEngineHeartbeatSender(
-            options,
-            windmillServer,
-            getDataStreamPool,
-            checkNotNull(configFetcher).getGlobalConfigHandle());
+    HeartbeatSender heartbeatSender = createStreamingEngineHeartbeatSender(windmillServer);
     @SuppressWarnings("methodref.receiver.bound")
     WorkCommitter workCommitter =
         StreamingEngineWorkCommitter.builder()
@@ -603,25 +597,9 @@ private static ChannelzServlet createChannelzServlet(
   }
 
   private static HeartbeatSender createStreamingEngineHeartbeatSender(
-      DataflowWorkerHarnessOptions options,
-      WindmillServerStub windmillClient,
-      WindmillStreamPool<GetDataStream> getDataStreamPool,
-      StreamingGlobalConfigHandle globalConfigHandle) {
-    // Experiment gates the logic till backend changes are rollback safe
-    if (!DataflowRunner.hasExperiment(
-            options, STREAMING_ENGINE_USE_JOB_SETTINGS_FOR_HEARTBEAT_POOL_EXPERIMENT)
-        || options.getUseSeparateWindmillHeartbeatStreams() != null) {
-      return StreamPoolHeartbeatSender.create(
-          Boolean.TRUE.equals(options.getUseSeparateWindmillHeartbeatStreams())
-              ? WindmillStreamPool.create(1, GET_DATA_STREAM_TIMEOUT, windmillClient::getDataStream)
-              : getDataStreamPool);
-
-    } else {
-      return StreamPoolHeartbeatSender.create(
-          WindmillStreamPool.create(1, GET_DATA_STREAM_TIMEOUT, windmillClient::getDataStream),
-          getDataStreamPool,
-          globalConfigHandle);
-    }
+      WindmillServerStub windmillClient) {
+    return StreamPoolHeartbeatSender.create(
+        WindmillStreamPool.create(1, GET_DATA_STREAM_TIMEOUT, windmillClient::getDataStream));
   }
 
   public static StreamingDataflowWorker fromOptions(DataflowWorkerHarnessOptions options) {
@@ -725,7 +703,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 +731,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 +898,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/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/refresh/StreamPoolHeartbeatSender.java

@@ -19,7 +19,6 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 import javax.annotation.Nonnull;
-import org.apache.beam.runners.dataflow.worker.streaming.config.StreamingGlobalConfigHandle;
 import org.apache.beam.runners.dataflow.worker.windmill.client.CloseableStream;
 import org.apache.beam.runners.dataflow.worker.windmill.client.WindmillStream;
 import org.apache.beam.runners.dataflow.worker.windmill.client.WindmillStreamPool;
@@ -47,32 +46,6 @@ public static StreamPoolHeartbeatSender create(
     return new StreamPoolHeartbeatSender(heartbeatStreamPool);
   }
 
-  /**
-   * Creates StreamPoolHeartbeatSender that switches between the passed in stream pools depending on
-   * global config.
-   *
-   * @param dedicatedHeartbeatPool stream to use when using separate streams for heartbeat is
-   *     enabled.
-   * @param getDataPool stream to use when using separate streams for heartbeat is disabled.
-   */
-  public static StreamPoolHeartbeatSender create(
-      @Nonnull WindmillStreamPool<WindmillStream.GetDataStream> dedicatedHeartbeatPool,
-      @Nonnull WindmillStreamPool<WindmillStream.GetDataStream> getDataPool,
-      @Nonnull StreamingGlobalConfigHandle configHandle) {
-    // Use getDataPool as the default, settings callback will
-    // switch to the separate pool if enabled before processing any elements are processed.
-    StreamPoolHeartbeatSender heartbeatSender = new StreamPoolHeartbeatSender(getDataPool);
-    configHandle.registerConfigObserver(
-        streamingGlobalConfig ->
-            heartbeatSender.heartbeatStreamPool.set(
-                streamingGlobalConfig
-                        .userWorkerJobSettings()
-                        .getUseSeparateWindmillHeartbeatStreams()
-                    ? dedicatedHeartbeatPool
-                    : getDataPool));
-    return heartbeatSender;
-  }
-
   @Override
   public void sendHeartbeats(Heartbeats heartbeats) {
     try (CloseableStream<WindmillStream.GetDataStream> closeableStream =