You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ah...@apache.org on 2025/04/30 14:01:59 UTC
(beam) branch master updated: [ManagedIO] Fail expansion when encountering extra or unknown configuration (#34525)
This is an automated email from the ASF dual-hosted git repository.
ahmedabualsaud 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 363bcc0d328 [ManagedIO] Fail expansion when encountering extra or unknown configuration (#34525)
363bcc0d328 is described below
commit 363bcc0d3286ccfb4f3ea39641d769c12c6d9eb8
Author: Ahmed Abualsaud <65...@users.noreply.github.com>
AuthorDate: Wed Apr 30 10:01:54 2025 -0400
[ManagedIO] Fail expansion when encountering extra or unknown configuration (#34525)
* fail expansion when encountering extra or unknown configuration
* add another test
* spotless
* add skip validation option
* tests
* new api instead of pipeline option
* extend to python
* trigger tests
* default to false
* fix
* remove test case not applicable for unit test
* address comments
* fix after merge HEAD
---
.../KafkaReadSchemaTransformProviderTest.java | 5 -
.../java/org/apache/beam/sdk/managed/Managed.java | 13 +++
.../managed/ManagedSchemaTransformProvider.java | 69 ++++++++++---
.../testing/TestSchemaTransformProvider.java | 13 ++-
.../ManagedSchemaTransformProviderTest.java | 112 ++++++++++++++++++++-
sdks/python/apache_beam/transforms/managed.py | 10 +-
.../transforms/managed_iceberg_it_test.py | 9 ++
7 files changed, 207 insertions(+), 24 deletions(-)
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProviderTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProviderTest.java
index a65583b5663..5ee8e76b2c4 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProviderTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProviderTest.java
@@ -304,11 +304,6 @@ public class KafkaReadSchemaTransformProviderTest {
+ "bootstrap_servers: some bootstrap\n"
+ "schema: '{\"type\":\"record\",\"name\":\"my_record\",\"fields\":[{\"name\":\"bool\",\"type\":\"boolean\"}]}'",
"topic: topic_3\n"
- + "bootstrap_servers: some bootstrap\n"
- + "schema_registry_url: some-url\n"
- + "schema_registry_subject: some-subject\n"
- + "format: RAW",
- "topic: topic_4\n"
+ "bootstrap_servers: some bootstrap\n"
+ "format: PROTO\n"
+ "schema: '"
diff --git a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java
index 3417c35233a..4496aae48b6 100644
--- a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java
+++ b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java
@@ -176,6 +176,8 @@ public class Managed {
@VisibleForTesting
abstract List<String> getSupportedIdentifiers();
+ abstract @Nullable Boolean getSkipConfigValidation();
+
abstract Builder toBuilder();
@AutoValue.Builder
@@ -189,6 +191,8 @@ public class Managed {
@VisibleForTesting
abstract Builder setSupportedIdentifiers(List<String> supportedIdentifiers);
+ abstract Builder setSkipConfigValidation(boolean skip);
+
abstract ManagedTransform build();
}
@@ -215,6 +219,14 @@ public class Managed {
return toBuilder().setSupportedIdentifiers(supportedIdentifiers).build();
}
+ /**
+ * Skips configuration validation. If unset, the pipeline will fail at construction time if the
+ * configuration includes unknown fields or missing required fields.
+ */
+ public ManagedTransform skipConfigValidation() {
+ return toBuilder().setSkipConfigValidation(true).build();
+ }
+
@Override
public PCollectionRowTuple expand(PInput input) {
PCollectionRowTuple inputTuple = resolveInput(input);
@@ -224,6 +236,7 @@ public class Managed {
.setTransformIdentifier(getIdentifier())
.setConfig(YamlUtils.yamlStringFromMap(getConfig()))
.setConfigUrl(getConfigUrl())
+ .setSkipConfigValidation(getSkipConfigValidation())
.build();
SchemaTransform underlyingTransform =
diff --git a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProvider.java b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProvider.java
index f4be3f0c9a6..9c7816ad539 100644
--- a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProvider.java
+++ b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProvider.java
@@ -26,15 +26,19 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
+import java.util.Set;
import javax.annotation.Nullable;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.fs.MatchResult;
+import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.schemas.AutoValueSchema;
import org.apache.beam.sdk.schemas.NoSuchSchemaException;
import org.apache.beam.sdk.schemas.Schema;
@@ -98,6 +102,11 @@ public class ManagedSchemaTransformProvider
@SchemaFieldDescription("YAML string config used to build the underlying SchemaTransform.")
public abstract @Nullable String getConfig();
+ @SchemaFieldDescription(
+ "Skips configuration validation. If unset, the pipeline will fail at construction "
+ + "time if the configuration includes unknown fields or missing required fields.")
+ public abstract @Nullable Boolean getSkipConfigValidation();
+
@AutoValue.Builder
public abstract static class Builder {
public abstract Builder setTransformIdentifier(String identifier);
@@ -106,6 +115,8 @@ public class ManagedSchemaTransformProvider
public abstract Builder setConfig(@Nullable String yamlConfig);
+ public abstract Builder setSkipConfigValidation(@Nullable Boolean skip);
+
public abstract ManagedConfig build();
}
@@ -154,28 +165,21 @@ public class ManagedSchemaTransformProvider
static class ManagedSchemaTransform extends SchemaTransform {
private final ManagedConfig managedConfig;
- private final Row underlyingRowConfig;
private final SchemaTransformProvider underlyingTransformProvider;
ManagedSchemaTransform(
ManagedConfig managedConfig, SchemaTransformProvider underlyingTransformProvider) {
- // parse config before expansion to check if it matches underlying transform's config schema
- Schema transformConfigSchema = underlyingTransformProvider.configurationSchema();
- Row underlyingRowConfig;
- try {
- underlyingRowConfig = getRowConfig(managedConfig, transformConfigSchema);
- } catch (Exception e) {
- throw new IllegalArgumentException(
- "Encountered an error when retrieving a configuration", e);
- }
-
- this.underlyingRowConfig = underlyingRowConfig;
this.underlyingTransformProvider = underlyingTransformProvider;
this.managedConfig = managedConfig;
}
@Override
public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ Row underlyingRowConfig =
+ getRowConfig(
+ managedConfig,
+ underlyingTransformProvider.configurationSchema(),
+ input.getPipeline().getOptions());
LOG.debug(
"Building transform \"{}\" with configuration: {}",
underlyingTransformProvider.identifier(),
@@ -184,6 +188,7 @@ public class ManagedSchemaTransformProvider
return input.apply(underlyingTransformProvider.from(underlyingRowConfig));
}
+ @VisibleForTesting
public ManagedConfig getManagedConfig() {
return this.managedConfig;
}
@@ -206,7 +211,8 @@ public class ManagedSchemaTransformProvider
// May return an empty row (perhaps the underlying transform doesn't have any required
// parameters)
@VisibleForTesting
- static Row getRowConfig(ManagedConfig config, Schema transformSchema) {
+ static Row getRowConfig(
+ ManagedConfig config, Schema transformConfigSchema, PipelineOptions options) {
Map<String, Object> configMap = config.resolveUnderlyingConfig();
// Build a config Row that will be used to build the underlying SchemaTransform.
// If a mapping for the SchemaTransform exists, we use it to update parameter names to align
@@ -225,7 +231,42 @@ public class ManagedSchemaTransformProvider
configMap = remappedConfig;
}
- return YamlUtils.toBeamRow(configMap, transformSchema, false);
+ @Nullable Boolean skipValidation = config.getSkipConfigValidation();
+ if (skipValidation == null || !skipValidation) {
+ validateUserConfig(
+ config.getTransformIdentifier(),
+ new HashSet<>(configMap.keySet()),
+ transformConfigSchema);
+ }
+
+ return YamlUtils.toBeamRow(configMap, transformConfigSchema, false);
+ }
+
+ static void validateUserConfig(
+ String transformId, Set<String> userParams, Schema transformConfigSchema) {
+ List<String> missingRequiredFields = new ArrayList<>();
+ for (Schema.Field field : transformConfigSchema.getFields()) {
+ boolean inUserConfig = userParams.remove(field.getName());
+ if (!field.getType().getNullable() && !inUserConfig) {
+ missingRequiredFields.add(field.getName());
+ }
+ }
+
+ if (!missingRequiredFields.isEmpty() || !userParams.isEmpty()) {
+ String msg = "Invalid config for transform '" + transformId + "':";
+ if (!missingRequiredFields.isEmpty()) {
+ msg += " Missing required fields: " + missingRequiredFields + ".";
+ }
+ if (!userParams.isEmpty()) {
+ msg +=
+ " Contains unknown fields: "
+ + userParams
+ + ". If you'd still like to pass "
+ + "these fields, use '.skipConfigValidation()'";
+ }
+
+ throw new IllegalArgumentException(msg);
+ }
}
public static Map<String, Map<String, String>> getAliases() {
diff --git a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/testing/TestSchemaTransformProvider.java b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/testing/TestSchemaTransformProvider.java
index 2771ecd0164..7a28050d475 100644
--- a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/testing/TestSchemaTransformProvider.java
+++ b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/testing/TestSchemaTransformProvider.java
@@ -17,6 +17,8 @@
*/
package org.apache.beam.sdk.managed.testing;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull;
+
import com.google.auto.service.AutoService;
import com.google.auto.value.AutoValue;
import org.apache.beam.sdk.schemas.AutoValueSchema;
@@ -31,6 +33,7 @@ import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.sdk.values.TypeDescriptors;
+import org.checkerframework.checker.nullness.qual.Nullable;
@AutoService(SchemaTransformProvider.class)
public class TestSchemaTransformProvider
@@ -52,19 +55,27 @@ public class TestSchemaTransformProvider
@SchemaFieldDescription("Integer to add to each row element.")
public abstract Integer getExtraInteger();
+ @SchemaFieldDescription("If true, will upper case the extra string. Default is false.")
+ public abstract @Nullable Boolean getToggleUppercase();
+
@AutoValue.Builder
public abstract static class Builder {
public abstract Builder setExtraString(String extraString);
public abstract Builder setExtraInteger(Integer extraInteger);
+ public abstract Builder setToggleUppercase(Boolean toggleUppercase);
+
public abstract Config build();
}
}
@Override
public SchemaTransform from(Config config) {
- String extraString = config.getExtraString();
+ String extraString =
+ firstNonNull(config.getToggleUppercase(), false)
+ ? config.getExtraString().toUpperCase()
+ : config.getExtraString();
Integer extraInteger = config.getExtraInteger();
return new SchemaTransform() {
@Override
diff --git a/sdks/java/managed/src/test/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProviderTest.java b/sdks/java/managed/src/test/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProviderTest.java
index a287ec6260c..a9565400333 100644
--- a/sdks/java/managed/src/test/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProviderTest.java
+++ b/sdks/java/managed/src/test/java/org/apache/beam/sdk/managed/ManagedSchemaTransformProviderTest.java
@@ -24,9 +24,16 @@ import static org.junit.Assert.assertTrue;
import java.net.URISyntaxException;
import java.nio.file.Paths;
import java.util.Arrays;
+import java.util.Map;
+import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.managed.testing.TestSchemaTransformProvider;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.utils.YamlUtils;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -36,6 +43,8 @@ import org.junit.runners.JUnit4;
@RunWith(JUnit4.class)
public class ManagedSchemaTransformProviderTest {
@Rule public transient ExpectedException thrown = ExpectedException.none();
+ private static final Schema EMPTY_SCHEMA = Schema.builder().build();
+ private static final Row EMPTY_ROW = Row.nullRow(EMPTY_SCHEMA);
@Test
public void testFailWhenNoConfigSpecified() {
@@ -49,6 +58,103 @@ public class ManagedSchemaTransformProviderTest {
config.validate();
}
+ @Test
+ public void testFailWhenUnknownFieldsSpecified() {
+ Map<String, Object> config =
+ ImmutableMap.of(
+ "extra_string",
+ "str",
+ "extra_integer",
+ 123,
+ "toggle_uppercase",
+ true,
+ "unknown_field",
+ "unknown");
+ ManagedSchemaTransformProvider.ManagedConfig managedConfig =
+ ManagedSchemaTransformProvider.ManagedConfig.builder()
+ .setTransformIdentifier(TestSchemaTransformProvider.IDENTIFIER)
+ .setConfig(YamlUtils.yamlStringFromMap(config))
+ .build();
+
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("Invalid config for transform");
+ thrown.expectMessage(TestSchemaTransformProvider.IDENTIFIER);
+ thrown.expectMessage("Contains unknown fields");
+ thrown.expectMessage("unknown_field");
+ Pipeline p = Pipeline.create();
+ new ManagedSchemaTransformProvider(null)
+ .from(managedConfig)
+ .expand(
+ PCollectionRowTuple.of(
+ "input", p.apply(Create.of(EMPTY_ROW).withRowSchema(EMPTY_SCHEMA))));
+ }
+
+ @Test
+ public void testFailWhenMissingRequiredFields() {
+ Map<String, Object> config = ImmutableMap.of("extra_string", "str", "toggle_uppercase", true);
+ ManagedSchemaTransformProvider.ManagedConfig managedConfig =
+ ManagedSchemaTransformProvider.ManagedConfig.builder()
+ .setTransformIdentifier(TestSchemaTransformProvider.IDENTIFIER)
+ .setConfig(YamlUtils.yamlStringFromMap(config))
+ .build();
+
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("Invalid config for transform");
+ thrown.expectMessage(TestSchemaTransformProvider.IDENTIFIER);
+ thrown.expectMessage("Missing required fields");
+ thrown.expectMessage("extra_integer");
+ Pipeline p = Pipeline.create();
+ new ManagedSchemaTransformProvider(null)
+ .from(managedConfig)
+ .expand(
+ PCollectionRowTuple.of(
+ "input", p.apply(Create.of(EMPTY_ROW).withRowSchema(EMPTY_SCHEMA))));
+ }
+
+ @Test
+ public void testPassWhenMissingNullableFields() {
+ Map<String, Object> config = ImmutableMap.of("extra_string", "str", "extra_integer", 123);
+ ManagedSchemaTransformProvider.ManagedConfig managedConfig =
+ ManagedSchemaTransformProvider.ManagedConfig.builder()
+ .setTransformIdentifier(TestSchemaTransformProvider.IDENTIFIER)
+ .setConfig(YamlUtils.yamlStringFromMap(config))
+ .build();
+
+ Pipeline p = Pipeline.create();
+ new ManagedSchemaTransformProvider(null)
+ .from(managedConfig)
+ .expand(
+ PCollectionRowTuple.of(
+ "input", p.apply(Create.of(EMPTY_ROW).withRowSchema(EMPTY_SCHEMA))));
+ }
+
+ @Test
+ public void testSkipConfigValidationWithUnknownFields() {
+ Map<String, Object> config =
+ ImmutableMap.of(
+ "extra_string",
+ "str",
+ "extra_integer",
+ 123,
+ "toggle_uppercase",
+ true,
+ "unknown_field",
+ "unknown");
+ ManagedSchemaTransformProvider.ManagedConfig managedConfig =
+ ManagedSchemaTransformProvider.ManagedConfig.builder()
+ .setTransformIdentifier(TestSchemaTransformProvider.IDENTIFIER)
+ .setConfig(YamlUtils.yamlStringFromMap(config))
+ .setSkipConfigValidation(true)
+ .build();
+
+ Pipeline p = Pipeline.create();
+ new ManagedSchemaTransformProvider(null)
+ .from(managedConfig)
+ .expand(
+ PCollectionRowTuple.of(
+ "input", p.apply(Create.of(EMPTY_ROW).withRowSchema(EMPTY_SCHEMA))));
+ }
+
@Test
public void testGetConfigRowFromYamlString() {
String yamlString = "extra_string: abc\n" + "extra_integer: 123";
@@ -65,7 +171,8 @@ public class ManagedSchemaTransformProviderTest {
.build();
Row returnedRow =
- ManagedSchemaTransformProvider.getRowConfig(config, TestSchemaTransformProvider.SCHEMA);
+ ManagedSchemaTransformProvider.getRowConfig(
+ config, TestSchemaTransformProvider.SCHEMA, PipelineOptionsFactory.create());
assertEquals(expectedRow, returnedRow);
}
@@ -88,7 +195,8 @@ public class ManagedSchemaTransformProviderTest {
.withFieldValue("extra_integer", 123)
.build();
Row configRow =
- ManagedSchemaTransformProvider.getRowConfig(config, TestSchemaTransformProvider.SCHEMA);
+ ManagedSchemaTransformProvider.getRowConfig(
+ config, TestSchemaTransformProvider.SCHEMA, PipelineOptionsFactory.create());
assertEquals(expectedRow, configRow);
}
diff --git a/sdks/python/apache_beam/transforms/managed.py b/sdks/python/apache_beam/transforms/managed.py
index 2a2054954e3..6113f953aed 100644
--- a/sdks/python/apache_beam/transforms/managed.py
+++ b/sdks/python/apache_beam/transforms/managed.py
@@ -112,6 +112,7 @@ class Read(PTransform):
source: str,
config: Optional[dict[str, Any]] = None,
config_url: Optional[str] = None,
+ skip_config_validation: bool = False,
expansion_service=None):
super().__init__()
self._source = source
@@ -126,6 +127,7 @@ class Read(PTransform):
self._underlying_identifier = identifier
self._yaml_config = yaml.dump(config)
self._config_url = config_url
+ self._skip_config_validation = skip_config_validation
def expand(self, input):
return input | SchemaAwareExternalTransform(
@@ -134,7 +136,8 @@ class Read(PTransform):
rearrange_based_on_discovery=True,
transform_identifier=self._underlying_identifier,
config=self._yaml_config,
- config_url=self._config_url)
+ config_url=self._config_url,
+ skip_config_validation=self._skip_config_validation)
def default_label(self) -> str:
return "Managed Read(%s)" % self._source.upper()
@@ -153,6 +156,7 @@ class Write(PTransform):
sink: str,
config: Optional[dict[str, Any]] = None,
config_url: Optional[str] = None,
+ skip_config_validation: bool = False,
expansion_service=None):
super().__init__()
self._sink = sink
@@ -167,6 +171,7 @@ class Write(PTransform):
self._underlying_identifier = identifier
self._yaml_config = yaml.dump(config)
self._config_url = config_url
+ self._skip_config_validation = skip_config_validation
def expand(self, input):
return input | SchemaAwareExternalTransform(
@@ -175,7 +180,8 @@ class Write(PTransform):
rearrange_based_on_discovery=True,
transform_identifier=self._underlying_identifier,
config=self._yaml_config,
- config_url=self._config_url)
+ config_url=self._config_url,
+ skip_config_validation=self._skip_config_validation)
def default_label(self) -> str:
return "Managed Write(%s)" % self._sink.upper()
diff --git a/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py b/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py
index 20cb52335c7..d402692199b 100644
--- a/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py
+++ b/sdks/python/apache_beam/transforms/managed_iceberg_it_test.py
@@ -75,7 +75,16 @@ class ManagedIcebergIT(unittest.TestCase):
| beam.managed.Read(beam.managed.ICEBERG, config=iceberg_config)
| beam.Map(lambda row: row._asdict()))
+ output_dicts_skipped_config = (
+ read_pipeline
+ | beam.managed.Read(
+ beam.managed.ICEBERG,
+ config=iceberg_config,
+ skip_config_validation=True)
+ | beam.Map(lambda row: row._asdict()))
+
assert_that(output_dicts, equal_to(expected_dicts))
+ assert_that(output_dicts_skipped_config, equal_to(expected_dicts))
if __name__ == '__main__':