[Spark][#36841] Reuse the legacy maxRecordsPerBatch option in the Structured Streaming runner

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

@@ -48,16 +48,13 @@ public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipe
 
   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();
+  // Mirrors the legacy SparkPipelineOptions declaration exactly, so users migrating from the
+  // legacy runner keep the same flag.
+  @Description("Max records per micro-batch. For streaming sources only.")
+  @Default.Long(-1)
+  Long getMaxRecordsPerBatch();
 
-  void setMaxRecordsPerMicroBatch(int value);
+  void setMaxRecordsPerBatch(Long maxRecordsPerBatch);
 
   @Description(
       "Maximum duration in milliseconds of a micro-batch trigger interval (streaming mode only).")

runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptionsTest.java

@@ -0,0 +1,57 @@
+/*
+ * 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.junit.Assert.assertEquals;
+
+import org.apache.beam.runners.spark.SparkPipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link SparkStructuredStreamingPipelineOptions}. */
+@RunWith(JUnit4.class)
+public class SparkStructuredStreamingPipelineOptionsTest {
+
+  /**
+   * {@code maxRecordsPerBatch} is declared by both {@link SparkPipelineOptions} and {@link
+   * SparkStructuredStreamingPipelineOptions} with identical signatures, so one options proxy serves
+   * both and the flag carries over when migrating between the runners.
+   */
+  @Test
+  public void maxRecordsPerBatchIsSharedWithLegacyOptions() {
+    PipelineOptions options = PipelineOptionsFactory.create();
+
+    SparkStructuredStreamingPipelineOptions streamingOptions =
+        options.as(SparkStructuredStreamingPipelineOptions.class);
+    assertEquals(Long.valueOf(-1), streamingOptions.getMaxRecordsPerBatch());
+
+    streamingOptions.setMaxRecordsPerBatch(500L);
+    assertEquals(Long.valueOf(500), options.as(SparkPipelineOptions.class).getMaxRecordsPerBatch());
+  }
+
+  @Test
+  public void maxRecordsPerBatchIsParsedFromArgs() {
+    SparkStructuredStreamingPipelineOptions options =
+        PipelineOptionsFactory.fromArgs("--maxRecordsPerBatch=42")
+            .as(SparkStructuredStreamingPipelineOptions.class);
+    assertEquals(Long.valueOf(42), options.getMaxRecordsPerBatch());
+  }
+}