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 b9061e1734cf..518525e53c62 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 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 static Set> getRegisteredOptions() { return Collections.unmodifiableSet(CACHE.get().registeredOptions); } + /** + * For internal use only; no backwards-compatibility guarantees. + * + *

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. + * + *

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; + + /** + * For internal use only; no backwards-compatibility guarantees. + * + *

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. + * + *

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 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 75796465b33e..57f0c8683ffa 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.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 @@ ComputedProperties updated( private final ImmutableMap 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 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 options) { this(bindOptions(options), Maps.newHashMap(), 0); } @@ -178,8 +199,17 @@ private static Map bindOptions(Map inputOpti private ProxyInvocationHandler( Map options, Map jsonOptions, int revision) { + this(options, jsonOptions, ImmutableSet.of(), revision); + } + + private ProxyInvocationHandler( + Map options, + Map jsonOptions, + Set 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 @@ public String getOptionName(Method method) { 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. + * + *

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 explicitlySetProperties() { + Set 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 @@ public void serialize(PipelineOptions value, JsonGenerator jgen, SerializerProvi jgen.writeObject(serializedDisplayData); jgen.writeNumberField("revision", handler.revision.get()); + if (Boolean.TRUE.equals(provider.getAttribute(SERIALIZE_DEFAULTS_ATTRIBUTE))) { + Set 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 @@ public PipelineOptions deserialize(JsonParser jp, DeserializationContext ctxt) } } int revision = objectNode.hasNonNull("revision") ? objectNode.get("revision").asInt() : 0; + Set 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 4dc9bca28640..b157e4ad785d 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 PipelineResult run(PipelineOptions options) { 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 45bcdc5c9aed..dabfec9259ca 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.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 void testDisplayDataIncludesExplicitlySetDefaults() { 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>() {}), + 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 8df2e0abf355..a79bfd739234 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.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 void testRunWithDummyEnvironmentVariableFails() { 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 { + 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 implements SerializableMatcher {