[pull] master from apache:master

runners/spark/4/build.gradle

@@ -53,12 +53,14 @@ tasks.validatesStructuredStreamingRunnerBatch {
 
 tasks.validatesRunner.dependsOn(validatesStructuredStreamingRunnerBatch)
 
-// Exclude DStream-based streaming tests from the shared-base copy: the Spark 4 module
-// supports only structured streaming (batch) and does not include legacy DStream support.
+// Exclude legacy DStream-based streaming tests (under runners/spark/translation/streaming)
+// from the shared-base copy: the Spark 4 module does not include legacy DStream support.
 // Streaming test utilities also depend on kafka.server.KafkaServerStartable which was
 // removed in Kafka 2.8.0 (the first Kafka version with a _2.13 artifact).
+// Note: structured streaming tests (**/structuredstreaming/translation/streaming/**) are
+// intentionally NOT excluded.
 tasks.named("copyTestSourceOverrides") {
-  exclude "**/translation/streaming/**"
+  exclude "**/runners/spark/translation/streaming/**"
 }
 
 // Spark 4 uses org.lz4:lz4-java instead of at.yawk.lz4:lz4-java

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java

@@ -39,4 +39,38 @@ public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipe
   boolean getUseActiveSparkSession();
 
   void setUseActiveSparkSession(boolean value);
+
+  @Description(
+      "Watermark delay in milliseconds applied to event timestamps of streaming sources "
+          + "(streaming mode only).")
+  @Default.Long(0)
+  long getWatermarkDelayMillis();
+
+  void setWatermarkDelayMillis(long value);
+
+  // Note: deliberately NOT named getMaxRecordsPerBatch. The legacy Spark runner's
+  // SparkPipelineOptions already declares Long getMaxRecordsPerBatch(); a same-name getter with a
+  // different return type breaks proxy generation for every registered PipelineOptions interface.
+  @Description(
+      "Maximum number of records to read per micro-batch from a streaming source "
+          + "(streaming mode only).")
+  @Default.Integer(1000)
+  int getMaxRecordsPerMicroBatch();
+
+  void setMaxRecordsPerMicroBatch(int value);
+
+  @Description(
+      "Maximum duration in milliseconds of a micro-batch trigger interval (streaming mode only).")
+  @Default.Long(500)
+  long getMaxBatchDurationMillis();
+
+  void setMaxBatchDurationMillis(long value);
+
+  @Description(
+      "Test-oriented: gracefully stop streaming queries after this many consecutive empty "
+          + "micro-batches. Disabled if negative (streaming mode only).")
+  @Default.Integer(-1)
+  int getStreamingStopAfterIdleBatches();
+
+  void setStreamingStopAfterIdleBatches(int value);
 }

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineResult.java

@@ -25,27 +25,33 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import javax.annotation.Nullable;
+import java.util.function.Supplier;
 import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.EvaluationContext;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.metrics.MetricResults;
 import org.apache.beam.sdk.util.UserCodeException;
 import org.apache.spark.SparkException;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Duration;
 
 public class SparkStructuredStreamingPipelineResult implements PipelineResult {
 
   private final Future<?> pipelineExecution;
+  // Supplies the context of the translated pipeline, null until translation has completed.
+  private final Supplier<? extends @Nullable EvaluationContext> evaluationContext;
   private final MetricsAccumulator metrics;
-  private @Nullable final Runnable onTerminalState;
+  private final @Nullable Runnable onTerminalState;
   private PipelineResult.State state;
 
   SparkStructuredStreamingPipelineResult(
       Future<?> pipelineExecution,
+      Supplier<? extends @Nullable EvaluationContext> evaluationContext,
       MetricsAccumulator metrics,
-      @Nullable final Runnable onTerminalState) {
+      final @Nullable Runnable onTerminalState) {
     this.pipelineExecution = pipelineExecution;
+    this.evaluationContext = evaluationContext;
     this.metrics = metrics;
     this.onTerminalState = onTerminalState;
     // pipelineExecution is expected to have started executing eagerly.
@@ -113,6 +119,10 @@ public MetricResults metrics() {
 
   @Override
   public PipelineResult.State cancel() throws IOException {
+    EvaluationContext ctx = evaluationContext.get();
+    if (ctx != null) {
+      ctx.stop();
+    }
     pipelineExecution.cancel(true);
     offerNewState(PipelineResult.State.CANCELLED);
     return state;

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunner.java

@@ -17,21 +17,20 @@
  */
 package org.apache.beam.runners.spark.structuredstreaming;
 
-import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
-
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicReference;
 import javax.annotation.Nullable;
 import org.apache.beam.runners.core.metrics.MetricsPusher;
 import org.apache.beam.runners.core.metrics.NoOpMetricsSink;
 import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
 import org.apache.beam.runners.spark.structuredstreaming.metrics.SparkBeamMetricSource;
 import org.apache.beam.runners.spark.structuredstreaming.translation.EvaluationContext;
 import org.apache.beam.runners.spark.structuredstreaming.translation.PipelineTranslator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.PipelineTranslatorFactory;
 import org.apache.beam.runners.spark.structuredstreaming.translation.SparkSessionFactory;
-import org.apache.beam.runners.spark.structuredstreaming.translation.batch.PipelineTranslatorBatch;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineRunner;
 import org.apache.beam.sdk.metrics.MetricsEnvironment;
@@ -54,9 +53,9 @@
  * href="https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html">Structured
  * Streaming framework</a>).
  *
- * <p><b>This runner is experimental, its coverage of the Beam model is still partial. Due to
- * limitations of the Structured Streaming framework (e.g. lack of support for multiple stateful
- * operators), streaming mode is not yet supported by this runner. </b>
+ * <p><b>This runner is experimental, its coverage of the Beam model is still partial. Streaming
+ * mode requires the Spark 4 module (beam-runners-spark-4); the shared Spark 3 module supports batch
+ * pipelines only. </b>
  *
  * <p>The runner translates transforms defined on a Beam pipeline to Spark `Dataset` transformations
  * (leveraging the high level Dataset API) and then submits these to Spark to be executed.
@@ -145,17 +144,26 @@ public SparkStructuredStreamingPipelineResult run(final Pipeline pipeline) {
             + " It is still experimental, its coverage of the Beam model is partial. ***");
 
     PipelineTranslator.detectStreamingMode(pipeline, options);
-    checkArgument(!options.isStreaming(), "Streaming is not supported.");
 
     final SparkSession sparkSession = SparkSessionFactory.getOrCreateSession(options);
     final MetricsAccumulator metrics = MetricsAccumulator.getInstance(sparkSession);
 
+    // Set once the pipeline is translated, so the result can stop an ongoing (streaming)
+    // evaluation on cancel. Remains null until translation completes.
+    final AtomicReference<EvaluationContext> ctxRef = new AtomicReference<>();
+
     final Future<?> submissionFuture =
-        runAsync(() -> translatePipeline(sparkSession, pipeline).evaluate());
+        runAsync(
+            () -> {
+              EvaluationContext ctx = translatePipeline(sparkSession, pipeline);
+              ctxRef.set(ctx);
+              ctx.evaluate();
+            });
 
     final SparkStructuredStreamingPipelineResult result =
         new SparkStructuredStreamingPipelineResult(
             submissionFuture,
+            ctxRef::get,
             metrics,
             sparkStopFn(sparkSession, options.getUseActiveSparkSession()));
 
@@ -186,7 +194,7 @@ private EvaluationContext translatePipeline(SparkSession sparkSession, Pipeline
 
     PipelineTranslator.replaceTransforms(pipeline, options);
 
-    PipelineTranslator pipelineTranslator = new PipelineTranslatorBatch();
+    PipelineTranslator pipelineTranslator = PipelineTranslatorFactory.create(options.isStreaming());
     return pipelineTranslator.translate(pipeline, sparkSession, options);
   }
 

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/EvaluationContext.java

@@ -40,10 +40,10 @@
  */
 @SuppressWarnings("Slf4jDoNotLogMessageOfExceptionExplicitly")
 @Internal
-public final class EvaluationContext {
+public class EvaluationContext {
   private static final Logger LOG = LoggerFactory.getLogger(EvaluationContext.class);
 
-  interface NamedDataset<T> {
+  public interface NamedDataset<T> {
     String name();
 
     @Nullable
@@ -53,11 +53,16 @@ interface NamedDataset<T> {
   private final Collection<? extends NamedDataset<?>> leaves;
   private final SparkSession session;
 
-  EvaluationContext(Collection<? extends NamedDataset<?>> leaves, SparkSession session) {
+  protected EvaluationContext(Collection<? extends NamedDataset<?>> leaves, SparkSession session) {
     this.leaves = leaves;
     this.session = session;
   }
 
+  /** The leaf datasets of the translated pipeline that require evaluation. */
+  protected Collection<? extends NamedDataset<?>> leaves() {
+    return leaves;
+  }
+
   /** Trigger evaluation of all leaf datasets. */
   public void evaluate() {
     for (NamedDataset<?> ds : leaves) {
@@ -113,6 +118,13 @@ public static <T> void evaluate(String name, Dataset<T> ds) {
     }
   }
 
+  /**
+   * Stops any ongoing streaming execution triggered by this context.
+   *
+   * <p>This is a no-op for batch pipelines.
+   */
+  public void stop() {}
+
   public SparkSession getSparkSession() {
     return session;
   }

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslator.java

@@ -25,6 +25,7 @@
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -128,7 +129,20 @@ public EvaluationContext translate(
     TranslatingVisitor translator = new TranslatingVisitor(session, options, dependencies.results);
     pipeline.traverseTopologically(translator);
 
-    return new EvaluationContext(translator.leaves, session);
+    return createEvaluationContext(translator.leaves, session, options);
+  }
+
+  /**
+   * Creates the {@link EvaluationContext} for the translated pipeline.
+   *
+   * <p>Subclasses may override this to return a specialized context, e.g. to evaluate streaming
+   * pipelines.
+   */
+  protected EvaluationContext createEvaluationContext(
+      Collection<? extends EvaluationContext.NamedDataset<?>> leaves,
+      SparkSession session,
+      SparkCommonPipelineOptions options) {
+    return new EvaluationContext(leaves, session);
   }
 
   /**
@@ -311,12 +325,14 @@ public <T> void putDataset(
       TranslationResult<?, T> result = getResult(pCollection);
       result.dataset = dataset;
 
-      if (cache && result.usages() > 1) {
+      // Caching and lineage breaking are batch-only optimizations, streaming datasets must pass
+      // through untouched.
+      if (cache && result.usages() > 1 && !dataset.isStreaming()) {
         LOG.info("Dataset {} will be cached for reuse.", result.name);
         dataset.persist(storageLevel); // use NONE to disable
       }
 
-      if (result.estimatePlanComplexity() > PLAN_COMPLEXITY_THRESHOLD) {
+      if (!dataset.isStreaming() && result.estimatePlanComplexity() > PLAN_COMPLEXITY_THRESHOLD) {
         // Break linage of dataset to limit planning overhead for complex query plans.
         LOG.info("Breaking linage of dataset {} to limit complexity of query plan.", result.name);
         result.dataset = sparkSession.createDataset(dataset.rdd(), dataset.encoder());

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslatorFactory.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation;
+
+import org.apache.beam.runners.spark.structuredstreaming.translation.batch.PipelineTranslatorBatch;
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Factory to create the {@link PipelineTranslator} matching the execution mode of the pipeline.
+ *
+ * <p>This shared base version only supports batch mode. The Spark 4 module shadows this file to
+ * additionally dispatch to a streaming translator.
+ */
+@Internal
+public final class PipelineTranslatorFactory {
+  private PipelineTranslatorFactory() {}
+
+  /** Creates a {@link PipelineTranslator} for the given execution mode. */
+  public static PipelineTranslator create(boolean streaming) {
+    if (streaming) {
+      throw new UnsupportedOperationException(
+          "Streaming pipelines require the Spark 4 runner (beam-runners-spark-4).");
+    }
+    return new PipelineTranslatorBatch();
+  }
+}

runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/SparkSessionFactory.java

@@ -175,6 +175,12 @@ private static SparkSession.Builder sessionBuilder(
       sparkConf.setIfMissing("spark.sql.shuffle.partitions", Integer.toString(partitions));
     }
 
+    // Spark 4 transformWithState (used for streaming pipelines) requires the RocksDB state store.
+    // This is a harmless, inert configuration for batch pipelines on Spark 3.
+    sparkConf.setIfMissing(
+        "spark.sql.streaming.stateStore.providerClass",
+        "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider");
+
     return SparkSession.builder().config(sparkConf);
   }
 

sdks/python/tox.ini

@@ -459,6 +459,7 @@ commands =
 [testenv:py310-tensorflow-212]
 deps =
   212:
+    googleapis-common-protos<1.70
     tensorflow>=2.12rc1,<2.13
     # Help pip resolve conflict with typing-extensions for old version of TF https://github.com/apache/beam/issues/30852
     pydantic<2.7