This is an automated email from the ASF dual-hosted git repository.
reuvenlax pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 046b0387e05 Merge pull request #40437 from
reuvenlax/merge_validates_runner_pipeline_options
046b0387e05 is described below
commit 046b0387e05f45ba37e529694fde248fbf7f667e
Author: Reuven Lax <[email protected]>
AuthorDate: Tue Oct 13 08:44:57 2026 -0700
Merge pull request #40437 from
reuvenlax/merge_validates_runner_pipeline_options
Add PipelineOptions changes for merging ValidatesRunner tests.
---
.../beam/sdk/options/PipelineOptionsFactory.java | 44 ++++++++++++
.../beam/sdk/options/ProxyInvocationHandler.java | 80 +++++++++++++++++++++-
.../org/apache/beam/sdk/testing/TestPipeline.java | 17 ++++-
.../sdk/options/ProxyInvocationHandlerTest.java | 71 +++++++++++++++++++
.../apache/beam/sdk/testing/TestPipelineTest.java | 49 +++++++++++++
5 files changed, 258 insertions(+), 3 deletions(-)
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
index b9061e1734c..518525e53c6 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
@@ -85,6 +85,7 @@ import javax.annotation.Nonnull;
import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionDescriptor;
import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionType;
import org.apache.beam.sdk.PipelineRunner;
+import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.runners.PipelineRunnerRegistrar;
import org.apache.beam.sdk.transforms.display.DisplayData;
@@ -575,6 +576,49 @@ public class PipelineOptionsFactory {
return Collections.unmodifiableSet(CACHE.get().registeredOptions);
}
+ /**
+ * <b><i>For internal use only; no backwards-compatibility
guarantees.</i></b>
+ *
+ * <p>Jackson serialization attribute (see {@link
+ * com.fasterxml.jackson.databind.ObjectWriter#withAttribute}) which, when
set to {@code true},
+ * makes the serialized form of a {@link PipelineOptions} additionally
record which of the
+ * serialized values were lazily bound defaults, so that {@link
#explicitlySetProperties} on the
+ * deserialized copy gives the same answer as on the original.
+ *
+ * <p>It is off by default: the serialized form handed to runners and
workers is unchanged, and a
+ * {@link PipelineOptions} deserialized from JSON without this marker
reports every serialized
+ * property as explicitly set.
+ */
+ @Internal
+ public static final String SERIALIZE_DEFAULTS_ATTRIBUTE =
+ ProxyInvocationHandler.SERIALIZE_DEFAULTS_ATTRIBUTE;
+
+ /**
+ * <b><i>For internal use only; no backwards-compatibility
guarantees.</i></b>
+ *
+ * <p>Returns the names of the properties of {@code options} that have been
explicitly set
+ * (through a setter, from command-line arguments, or from the JSON the
object was deserialized
+ * from), as opposed to defaults that were lazily bound the first time a
getter was called.
+ *
+ * <p>Unlike the serialized form of {@code options}, which contains every
bound value, this allows
+ * callers to compare two options objects by what was actually configured on
them without being
+ * confused by which getters happen to have been invoked on each. Note that
a serialization round
+ * trip loses this distinction unless the producer set {@link
#SERIALIZE_DEFAULTS_ATTRIBUTE}.
+ *
+ * @throws IllegalArgumentException if {@code options} was not created by
this factory
+ */
+ @Internal
+ public static Set<String> explicitlySetProperties(PipelineOptions options) {
+ checkArgumentNotNull(options);
+ checkArgument(
+ Proxy.isProxyClass(options.getClass())
+ && Proxy.getInvocationHandler(options) instanceof
ProxyInvocationHandler,
+ "PipelineOptions %s was not created by PipelineOptionsFactory",
+ options.getClass().getName());
+ return Collections.unmodifiableSet(
+ ((ProxyInvocationHandler)
Proxy.getInvocationHandler(options)).explicitlySetProperties());
+ }
+
/**
* Outputs the set of registered options with the PipelineOptionsFactory
with a description for
* each one if available to the output stream. This output is pretty printed
and meant to be human
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java
index 75796465b33..57f0c8683ff 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java
@@ -53,6 +53,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
+import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
@@ -163,6 +164,26 @@ class ProxyInvocationHandler implements InvocationHandler,
Serializable {
private final ImmutableMap<String, JsonNode> jsonOptions;
+ /**
+ * Names of the entries in {@link #jsonOptions} that were lazily bound
defaults, rather than
+ * explicitly set values, on the instance that was serialized. Only
populated when that instance
+ * was serialized with {@link #SERIALIZE_DEFAULTS_ATTRIBUTE}; otherwise
empty, in which case every
+ * entry of {@link #jsonOptions} is treated as explicitly set.
+ */
+ private final ImmutableSet<String> jsonDefaults;
+
+ /**
+ * Jackson serialization attribute which, when set to {@code true} on the
{@link
+ * com.fasterxml.jackson.databind.ObjectWriter}, makes {@link Serializer}
additionally record
+ * which serialized values were lazily bound defaults, so that {@link
#explicitlySetProperties()}
+ * on the deserialized instance can tell them apart from explicitly set
values. Off by default so
+ * that the serialized form seen by runners and workers is unchanged.
+ */
+ static final String SERIALIZE_DEFAULTS_ATTRIBUTE =
+ "org.apache.beam.sdk.options.ProxyInvocationHandler.serializeDefaults";
+
+ private static final String DEFAULTS_FIELD = "defaults";
+
ProxyInvocationHandler(Map<String, Object> options) {
this(bindOptions(options), Maps.newHashMap(), 0);
}
@@ -178,8 +199,17 @@ class ProxyInvocationHandler implements InvocationHandler,
Serializable {
private ProxyInvocationHandler(
Map<String, BoundValue> options, Map<String, JsonNode> jsonOptions, int
revision) {
+ this(options, jsonOptions, ImmutableSet.of(), revision);
+ }
+
+ private ProxyInvocationHandler(
+ Map<String, BoundValue> options,
+ Map<String, JsonNode> jsonOptions,
+ Set<String> jsonDefaults,
+ int revision) {
this.options = new ConcurrentHashMap<>(options);
this.jsonOptions = ImmutableMap.copyOf(jsonOptions);
+ this.jsonDefaults = ImmutableSet.copyOf(jsonDefaults);
this.revision = new AtomicInteger(revision);
this.computedProperties =
new ComputedProperties(
@@ -249,6 +279,34 @@ class ProxyInvocationHandler implements InvocationHandler,
Serializable {
return computedProperties.gettersToPropertyNames.get(method.getName());
}
+ /**
+ * Returns the names of the properties that have been explicitly set, either
through a setter or
+ * by construction from arguments or JSON, as opposed to defaults that were
lazily bound by a
+ * getter.
+ *
+ * <p>Properties present in the JSON this instance was deserialized from are
counted as explicitly
+ * set unless that JSON was produced with {@link
#SERIALIZE_DEFAULTS_ATTRIBUTE} and marked them as
+ * defaults; without that marker the serialized form does not record whether
a value was a default
+ * on the instance that was serialized.
+ */
+ Set<String> explicitlySetProperties() {
+ Set<String> result = new HashSet<>();
+ for (String name : Sets.union(jsonOptions.keySet(), options.keySet())) {
+ if (isExplicitlySet(name)) {
+ result.add(name);
+ }
+ }
+ return result;
+ }
+
+ private boolean isExplicitlySet(String name) {
+ BoundValue bound = options.get(name);
+ if (bound != null && !bound.isDefault()) {
+ return true;
+ }
+ return jsonOptions.containsKey(name) && !jsonDefaults.contains(name);
+ }
+
private void writeObject(java.io.ObjectOutputStream stream) throws
IOException {
throw new NotSerializableException(
"PipelineOptions objects are not serializable and should not be
embedded into transforms "
@@ -810,6 +868,16 @@ class ProxyInvocationHandler implements InvocationHandler,
Serializable {
jgen.writeObject(serializedDisplayData);
jgen.writeNumberField("revision", handler.revision.get());
+ if
(Boolean.TRUE.equals(provider.getAttribute(SERIALIZE_DEFAULTS_ATTRIBUTE))) {
+ Set<String> defaults = new TreeSet<>();
+ for (String name : serializableOptions.keySet()) {
+ if (!handler.isExplicitlySet(name)) {
+ defaults.add(name);
+ }
+ }
+ jgen.writeFieldName(DEFAULTS_FIELD);
+ jgen.writeObject(defaults);
+ }
jgen.writeEndObject();
}
@@ -909,8 +977,18 @@ class ProxyInvocationHandler implements InvocationHandler,
Serializable {
}
}
int revision = objectNode.hasNonNull("revision") ?
objectNode.get("revision").asInt() : 0;
+ Set<String> defaults = new HashSet<>();
+ JsonNode defaultsNode = objectNode.get(DEFAULTS_FIELD);
+ if (defaultsNode != null && defaultsNode.isArray()) {
+ for (JsonNode name : defaultsNode) {
+ if (name.isTextual() && fields.containsKey(name.asText())) {
+ defaults.add(name.asText());
+ }
+ }
+ }
PipelineOptions options =
- new ProxyInvocationHandler(Maps.newHashMap(), fields,
revision).as(PipelineOptions.class);
+ new ProxyInvocationHandler(Maps.newHashMap(), fields, defaults,
revision)
+ .as(PipelineOptions.class);
ValueProvider.RuntimeValueProvider.setRuntimeOptions(options);
return options;
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
index 4dc9bca2864..b157e4ad785 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
@@ -434,8 +434,21 @@ public class TestPipeline extends Pipeline implements
TestRule {
final PipelineResult pipelineResult;
try {
enforcement.get().beforePipelineExecution();
- PipelineOptions updatedOptions =
- MAPPER.convertValue(MAPPER.valueToTree(options),
PipelineOptions.class);
+ // Round-trip through JSON so that the pipeline runs on a private copy
of the options. The
+ // defaults marker keeps PipelineOptionsFactory.explicitlySetProperties
meaningful on that
+ // copy; it is only ever set here and does not reach the serialized form
of a submitted job.
+ PipelineOptions updatedOptions;
+ try {
+ updatedOptions =
+ MAPPER.readValue(
+ MAPPER
+ .writer()
+
.withAttribute(PipelineOptionsFactory.SERIALIZE_DEFAULTS_ATTRIBUTE, true)
+ .writeValueAsBytes(options),
+ PipelineOptions.class);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(e.getMessage(), e);
+ }
updatedOptions
.as(TestValueProviderOptions.class)
.setProviderRuntimeValues(StaticValueProvider.of(providerRuntimeValues));
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
index 45bcdc5c9ae..dabfec9259c 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
@@ -39,6 +39,7 @@ import com.fasterxml.jackson.annotation.JsonValue;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.JsonNode;
@@ -1111,6 +1112,76 @@ public class ProxyInvocationHandlerTest {
assertThat(data, hasDisplayItem("foo", defaultValue));
}
+ @Test
+ public void testExplicitlySetPropertiesIgnoresLazilyBoundDefaults() {
+ HasDefaults options = PipelineOptionsFactory.as(HasDefaults.class);
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(options),
not(hasItem("foo")));
+
+ // Reading a default binds it, but does not make it explicitly set.
+ assertEquals("bar", options.getFoo());
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(options),
not(hasItem("foo")));
+
+ // Setting it, even to the default value, does.
+ options.setFoo("bar");
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(options),
hasItem("foo"));
+ }
+
+ @Test
+ public void testExplicitlySetPropertiesIncludesArgumentsAndJsonOptions()
throws Exception {
+ FooOptions options =
PipelineOptionsFactory.fromArgs("--foo=baz").as(FooOptions.class);
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(options),
hasItem("foo"));
+
+ FooOptions deserialized = serializeDeserialize(FooOptions.class, options);
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized),
hasItem("foo"));
+ }
+
+ @Test
+ public void
testSerializationWithoutDefaultsMarkerIsUnchangedAndCountsAllAsExplicit()
+ throws Exception {
+ HasDefaults options = PipelineOptionsFactory.as(HasDefaults.class);
+ assertEquals("bar", options.getFoo()); // binds the default
+
+ JsonNode serialized = MAPPER.readTree(MAPPER.writeValueAsString(options));
+ assertTrue(serialized.has("options"));
+ assertTrue(serialized.has("display_data"));
+ assertTrue(serialized.has("revision"));
+ assertFalse("defaults marker must be opt-in", serialized.has("defaults"));
+
+ HasDefaults deserialized = serializeDeserialize(HasDefaults.class,
options);
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized),
hasItem("foo"));
+ }
+
+ @Test
+ public void testSerializationWithDefaultsMarkerPreservesExplicitness()
throws Exception {
+ HasDefaults options =
PipelineOptionsFactory.fromArgs("--foo=explicit").as(HasDefaults.class);
+ options.as(ObjectPipelineOptions.class).getValue(); // binds a default
+
+ byte[] serialized =
+ MAPPER
+ .writer()
+
.withAttribute(PipelineOptionsFactory.SERIALIZE_DEFAULTS_ATTRIBUTE, true)
+ .writeValueAsBytes(options);
+ JsonNode tree = MAPPER.readTree(serialized);
+ assertTrue(tree.get("options").has("value"));
+ assertThat(
+ MAPPER.convertValue(tree.get("defaults"), new
TypeReference<List<String>>() {}),
+ hasItem("value"));
+
+ PipelineOptions deserialized = MAPPER.readValue(serialized,
PipelineOptions.class);
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized),
hasItem("foo"));
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized),
not(hasItem("value")));
+ // The marked default is still served from the JSON.
+ assertNull(deserialized.as(ObjectPipelineOptions.class).getValue());
+
+ // Setting the property on the copy makes it explicit.
+ deserialized.as(ObjectPipelineOptions.class).setValue("set");
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized),
hasItem("value"));
+
+ // A further serialization without the attribute drops the marker again.
+ JsonNode reserialized =
MAPPER.readTree(MAPPER.writeValueAsString(deserialized));
+ assertFalse(reserialized.has("defaults"));
+ }
+
@Test
public void testDisplayDataNullValuesConvertedToEmptyString() throws
Exception {
FooOptions options = PipelineOptionsFactory.as(FooOptions.class);
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestPipelineTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestPipelineTest.java
index 8df2e0abf35..a79bfd73923 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestPipelineTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestPipelineTest.java
@@ -18,10 +18,14 @@
package org.apache.beam.sdk.testing;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.sameInstance;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.Serializable;
@@ -33,9 +37,12 @@ import java.util.List;
import java.util.UUID;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.PipelineRunner;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.options.ApplicationNameOptions;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.MapElements;
@@ -121,6 +128,48 @@ public class TestPipelineTest implements Serializable {
pipeline.run();
}
+ @Test
+ public void testRunPreservesWhichOptionsWereExplicitlySet() {
+ pipeline.getOptions().setRunner(OptionsCapturingRunner.class);
+ // Reading a default binds it on the options object without making it
explicitly set.
+
assertFalse(pipeline.getOptions().as(StreamingOptions.class).isStreaming());
+ pipeline.apply(Create.of(1, 2, 3));
+
+ OptionsCapturingRunner.captured = null;
+ try {
+ pipeline.run();
+ fail("expected the runner to throw");
+ } catch (UnsupportedOperationException expected) {
+ // The runner always throws; what matters is the options it was
created with.
+ }
+
+ PipelineOptions seenByRunner = OptionsCapturingRunner.captured;
+ assertNotNull(seenByRunner);
+ assertThat(seenByRunner, not(sameInstance(pipeline.getOptions())));
+ // The copy still serves the bound default ...
+ assertFalse(seenByRunner.as(StreamingOptions.class).isStreaming());
+ // ... but does not report it as explicitly set, while explicitly set
options are reported.
+ assertThat(
+ PipelineOptionsFactory.explicitlySetProperties(seenByRunner),
not(hasItem("streaming")));
+ assertThat(PipelineOptionsFactory.explicitlySetProperties(seenByRunner),
hasItem("runner"));
+ }
+
+ /** A runner that records the options it was constructed from and refuses
to run. */
+ public static class OptionsCapturingRunner extends
PipelineRunner<PipelineResult> {
+ static volatile @Nullable PipelineOptions captured;
+
+ @SuppressWarnings("unused") // used by reflection
+ public static OptionsCapturingRunner fromOptions(PipelineOptions
options) {
+ captured = options;
+ return new OptionsCapturingRunner();
+ }
+
+ @Override
+ public PipelineResult run(Pipeline pipeline) {
+ throw new UnsupportedOperationException("OptionsCapturingRunner does
not run pipelines");
+ }
+ }
+
/** TestMatcher is a matcher designed for testing matcher
serialization/deserialization. */
public static class TestMatcher extends BaseMatcher<PipelineResult>
implements SerializableMatcher<PipelineResult> {
