feat: add Spark 4 structured streaming runner source

Add the Spark 4 structured streaming runner implementation and tests.
Most files are adapted from the Spark 3 structured streaming runner
with targeted changes for Spark 4 / Scala 2.13 API compatibility.

Key Spark 4-specific changes (diff against runners/spark/3/src/):

EncoderFactory — Replaced the direct ExpressionEncoder constructor
  (removed in Spark 4) with BeamAgnosticEncoder, a named class
  implementing both AgnosticExpressionPathEncoder (for expression
  delegation via toCatalyst/fromCatalyst) and AgnosticEncoders
  .StructEncoder (so Dataset.select(TypedColumn) creates an N-attribute
  plan, preventing FIELD_NUMBER_MISMATCH). The toCatalyst/fromCatalyst
  methods substitute the provided input expression via transformUp,
  enabling correct nesting inside composite encoders like
  Encoders.tuple().

EncoderHelpers — Added toExpressionEncoder() helper to handle Spark 4
  built-in encoders that are AgnosticEncoder subclasses rather than
  ExpressionEncoder.

GroupByKeyTranslatorBatch — Migrated from internal catalyst Expression
  API (CreateNamedStruct, Literal$) to public Column API (struct(),
  lit(), array()), as required by Spark 4.

BoundedDatasetFactory — Use classic.Dataset$.MODULE$.ofRows() as
  Dataset moved to org.apache.spark.sql.classic in Spark 4.

ScalaInterop — Replace WrappedArray.ofRef (removed in Scala 2.13)
  with JavaConverters.asScalaBuffer().toList() in seqOf().

GroupByKeyHelpers, CombinePerKeyTranslatorBatch — Replace
  TraversableOnce with IterableOnce (Scala 2.13 rename).

SparkStructuredStreamingPipelineResult — Replace sparkproject.guava
  with Beam's vendored Guava.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: tkaymak

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

@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.PipelineOptions;
+
+/**
+ * Spark runner {@link PipelineOptions} handles Spark execution-related configurations, such as the
+ * master address, and other user-related knobs.
+ */
+public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {
+
+  /** Set to true to run the job in test mode. */
+  @Default.Boolean(false)
+  boolean getTestMode();
+
+  void setTestMode(boolean testMode);
+
+  @Description("Enable if the runner should use the currently active Spark session.")
+  @Default.Boolean(false)
+  boolean getUseActiveSparkSession();
+
+  void setUseActiveSparkSession(boolean value);
+}

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

@@ -0,0 +1,131 @@
+/*
+ * 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;
+
+import static org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
+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.joda.time.Duration;
+
+public class SparkStructuredStreamingPipelineResult implements PipelineResult {
+
+  private final Future<?> pipelineExecution;
+  private final MetricsAccumulator metrics;
+  private @Nullable final Runnable onTerminalState;
+  private PipelineResult.State state;
+
+  SparkStructuredStreamingPipelineResult(
+      Future<?> pipelineExecution,
+      MetricsAccumulator metrics,
+      @Nullable final Runnable onTerminalState) {
+    this.pipelineExecution = pipelineExecution;
+    this.metrics = metrics;
+    this.onTerminalState = onTerminalState;
+    // pipelineExecution is expected to have started executing eagerly.
+    this.state = State.RUNNING;
+  }
+
+  private static RuntimeException runtimeExceptionFrom(final Throwable e) {
+    return (e instanceof RuntimeException) ? (RuntimeException) e : new RuntimeException(e);
+  }
+
+  /**
+   * Unwrap cause of SparkException or UserCodeException as PipelineExecutionException. Otherwise,
+   * return {@code exception} as RuntimeException.
+   */
+  private static RuntimeException unwrapCause(Throwable exception) {
+    Throwable next = exception;
+    while (next != null && (next instanceof SparkException || next instanceof UserCodeException)) {
+      exception = next;
+      next = next.getCause();
+    }
+    return exception == next
+        ? runtimeExceptionFrom(exception)
+        : new Pipeline.PipelineExecutionException(firstNonNull(next, exception));
+  }
+
+  private State awaitTermination(Duration duration)
+      throws TimeoutException, ExecutionException, InterruptedException {
+    pipelineExecution.get(duration.getMillis(), TimeUnit.MILLISECONDS);
+    // Throws an exception if the job is not finished successfully in the given time.
+    return PipelineResult.State.DONE;
+  }
+
+  @Override
+  public PipelineResult.State getState() {
+    return state;
+  }
+
+  @Override
+  public PipelineResult.State waitUntilFinish() {
+    return waitUntilFinish(Duration.millis(Long.MAX_VALUE));
+  }
+
+  @Override
+  public State waitUntilFinish(final Duration duration) {
+    try {
+      State finishState = awaitTermination(duration);
+      offerNewState(finishState);
+    } catch (final TimeoutException e) {
+      // ignore.
+    } catch (final ExecutionException e) {
+      offerNewState(PipelineResult.State.FAILED);
+      throw unwrapCause(firstNonNull(e.getCause(), e));
+    } catch (final Exception e) {
+      offerNewState(PipelineResult.State.FAILED);
+      throw unwrapCause(e);
+    }
+
+    return state;
+  }
+
+  @Override
+  public MetricResults metrics() {
+    return asAttemptedOnlyMetricResults(metrics.value());
+  }
+
+  @Override
+  public PipelineResult.State cancel() throws IOException {
+    offerNewState(PipelineResult.State.CANCELLED);
+    return state;
+  }
+
+  private void offerNewState(State newState) {
+    State oldState = this.state;
+    this.state = newState;
+    if (!oldState.isTerminal() && newState.isTerminal() && onTerminalState != null) {
+      try {
+        onTerminalState.run();
+      } catch (Exception e) {
+        throw unwrapCause(e);
+      }
+    }
+  }
+}