diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java index 29cc4cb99cfb..391350fd348e 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java +++ b/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).") diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptionsTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptionsTest.java new file mode 100644 index 000000000000..48c13f0ad577 --- /dev/null +++ b/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()); + } +}