You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by sh...@apache.org on 2026/07/22 20:20:27 UTC
(beam) branch fadvice created (now 19e66b9f9ed)
This is an automated email from the ASF dual-hosted git repository.
shunping pushed a change to branch fadvice
in repository https://gitbox.apache.org/repos/asf/beam.git
at 19e66b9f9ed Add a unit test
This branch includes the following new commits:
new 064aa6ca086 Support serialization and deserialization for GoogleCloudStorageReadOptions
new 19e66b9f9ed Add a unit test
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 02/02: Add a unit test
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch fadvice
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 19e66b9f9edbbe0b6afe8e31e41eebc7ea34f03b
Author: Shunping Huang <sh...@google.com>
AuthorDate: Wed Jul 22 16:16:03 2026 -0400
Add a unit test
---
.../sdk/extensions/gcp/options/GcsOptionsTest.java | 31 ++++++++++++++++++++++
1 file changed, 31 insertions(+)
diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptionsTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptionsTest.java
index c499290b851..de460ca0986 100644
--- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptionsTest.java
+++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptionsTest.java
@@ -18,11 +18,16 @@
package org.apache.beam.sdk.extensions.gcp.options;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThrows;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadOptions;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -75,4 +80,30 @@ public class GcsOptionsTest {
IllegalArgumentException.class,
() -> PipelineOptionsFactory.fromArgs(TOO_MANY_ENTRIES_WITH_JOB).as(GcsOptions.class));
}
+
+ @Test
+ public void testGoogleCloudStorageReadOptionsSerialization() throws Exception {
+ GcsOptions options = PipelineOptionsFactory.as(GcsOptions.class);
+ GoogleCloudStorageReadOptions readOptions =
+ GoogleCloudStorageReadOptions.builder()
+ .setFadvise(GoogleCloudStorageReadOptions.Fadvise.SEQUENTIAL)
+ .setFastFailOnNotFoundEnabled(false)
+ .setMinRangeRequestSize(12345L)
+ .build();
+ options.setGoogleCloudStorageReadOptions(readOptions);
+
+ ObjectMapper mapper = new ObjectMapper();
+ String serialized = mapper.writeValueAsString(options);
+ GcsOptions deserialized =
+ mapper.readValue(serialized, PipelineOptions.class).as(GcsOptions.class);
+
+ GoogleCloudStorageReadOptions deserializedReadOptions =
+ deserialized.getGoogleCloudStorageReadOptions();
+
+ assertNotNull(deserializedReadOptions);
+ assertEquals(
+ GoogleCloudStorageReadOptions.Fadvise.SEQUENTIAL, deserializedReadOptions.getFadvise());
+ assertFalse(deserializedReadOptions.isFastFailOnNotFoundEnabled());
+ assertEquals(12345L, deserializedReadOptions.getMinRangeRequestSize());
+ }
}
(beam) 01/02: Support serialization and deserialization for GoogleCloudStorageReadOptions
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch fadvice
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 064aa6ca086a1d877c72d91db183901af059256d
Author: Shunping Huang <sh...@google.com>
AuthorDate: Wed Jul 22 16:12:38 2026 -0400
Support serialization and deserialization for GoogleCloudStorageReadOptions
Fixes an issue where GoogleCloudStorageReadOptions set on GcsOptions
was annotated with @JsonIgnore, causing it to be omitted when serializing
PipelineOptions for remote workers (e.g., Dataflow) and falling back to
defaults.
---
.../sdk/extensions/gcp/options/GcsOptions.java | 53 +++++++++++++++++++++-
.../sdk/extensions/gcp/GcpCoreApiSurfaceTest.java | 2 +
2 files changed, 54 insertions(+), 1 deletion(-)
diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java
index 2da382a5b67..35cba02f7c1 100644
--- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java
+++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java
@@ -18,8 +18,19 @@
package org.apache.beam.sdk.extensions.gcp.options;
import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadOptions;
import com.google.cloud.hadoop.util.AsyncWriteChannelOptions;
+import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import org.apache.beam.sdk.extensions.gcp.storage.GcsPathValidator;
@@ -53,8 +64,48 @@ public interface GcsOptions extends ApplicationNameOptions, GcpOptions, Pipeline
}
}
+ class GcsReadOptionsSerializer extends JsonSerializer<GoogleCloudStorageReadOptions> {
+ @Override
+ public void serialize(
+ GoogleCloudStorageReadOptions value, JsonGenerator gen, SerializerProvider serializers)
+ throws java.io.IOException {
+ serializers.defaultSerializeValue(value, gen);
+ }
+ }
+
+ class GcsReadOptionsDeserializer extends JsonDeserializer<GoogleCloudStorageReadOptions> {
+ @Override
+ public GoogleCloudStorageReadOptions deserialize(JsonParser p, DeserializationContext ctxt)
+ throws java.io.IOException {
+ ObjectMapper mapper = (ObjectMapper) p.getCodec();
+ JsonNode root = mapper.readTree(p);
+ GoogleCloudStorageReadOptions.Builder builder = GoogleCloudStorageReadOptions.builder();
+
+ for (Method method : GoogleCloudStorageReadOptions.Builder.class.getMethods()) {
+ if (method.getName().startsWith("set") && method.getParameterCount() == 1) {
+ String propName =
+ Character.toLowerCase(method.getName().charAt(3)) + method.getName().substring(4);
+ JsonNode node = root.get(propName);
+ if (node != null && !node.isNull()) {
+ try {
+ Class<?> paramType = method.getParameterTypes()[0];
+ Object val = mapper.treeToValue(node, paramType);
+ if (val != null) {
+ method.invoke(builder, java.util.Objects.requireNonNull(val));
+ }
+ } catch (Exception ignored) {
+ // Ignore incompatible or unmappable properties
+ }
+ }
+ }
+ }
+ return builder.build();
+ }
+ }
+
/** @deprecated This option will be removed in a future release. */
- @JsonIgnore
+ @JsonSerialize(using = GcsReadOptionsSerializer.class)
+ @JsonDeserialize(using = GcsReadOptionsDeserializer.class)
@Description(
"The GoogleCloudStorageReadOptions instance that should be used to read from Google Cloud Storage.")
@Default.InstanceFactory(GcsReadOptionsFactory.class)
diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/GcpCoreApiSurfaceTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/GcpCoreApiSurfaceTest.java
index 8af5e2260fc..cd51d4fd9d2 100644
--- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/GcpCoreApiSurfaceTest.java
+++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/GcpCoreApiSurfaceTest.java
@@ -51,6 +51,8 @@ public class GcpCoreApiSurfaceTest {
final Set<Matcher<Class<?>>> allowedClasses =
ImmutableSet.of(
classesInPackage("com.fasterxml.jackson.annotation"),
+ classesInPackage("com.fasterxml.jackson.core"),
+ classesInPackage("com.fasterxml.jackson.databind"),
classesInPackage("com.google.api.client.googleapis"),
classesInPackage("com.google.api.client.http"),
classesInPackage("com.google.api.client.json"),