diff --git a/sdks/java/extensions/google-cloud-platform-core/build.gradle b/sdks/java/extensions/google-cloud-platform-core/build.gradle index f1bfb63c7a3a..78cfe4739ec5 100644 --- a/sdks/java/extensions/google-cloud-platform-core/build.gradle +++ b/sdks/java/extensions/google-cloud-platform-core/build.gradle @@ -56,6 +56,7 @@ dependencies { implementation library.java.http_core implementation library.java.http_client implementation library.java.jackson_annotations + implementation library.java.jackson_core implementation library.java.jackson_databind permitUnusedDeclared library.java.jackson_databind // BEAM-11761 testImplementation project(path: ":sdks:java:core", configuration: "shadowTest") 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 2da382a5b674..134c4cb3f281 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,6 +18,15 @@ 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.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.util.HashMap; @@ -49,12 +58,16 @@ public interface GcsOptions extends ApplicationNameOptions, GcpOptions, Pipeline class GcsReadOptionsFactory implements DefaultValueFactory { @Override public GoogleCloudStorageReadOptions create(PipelineOptions options) { - return GoogleCloudStorageReadOptions.DEFAULT; + // In gcs-connector v3, GoogleCloudStorageReadOptions.DEFAULT changed fadvise from SEQUENTIAL + // to AUTO. Beam workloads default to SEQUENTIAL to preserve expected sequential read + // throughput and caching behavior. + return GcsReadOptionsSerializer.DEFAULT_OPTIONS; } } /** @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) @@ -286,3 +299,71 @@ boolean exceedsEntryLimit() { } } } + +class GcsReadOptionsSerializer extends JsonSerializer { + static final GoogleCloudStorageReadOptions DEFAULT_OPTIONS = + GoogleCloudStorageReadOptions.DEFAULT + .toBuilder() + .setFadvise(GoogleCloudStorageReadOptions.Fadvise.SEQUENTIAL) + .build(); + + @Override + public void serialize( + GoogleCloudStorageReadOptions value, JsonGenerator gen, SerializerProvider serializers) + throws java.io.IOException { + // Note: We only support a partial set of options to propagate to remote + // workers. Setting the unsupported ones will not have any effect and will fall + // back to default in remote workers. Support for additional options can be + // added on a need basis. The full list of options can be seen in + // com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadOptions. + gen.writeStartObject(); + if (value.getFadvise() != null && value.getFadvise() != DEFAULT_OPTIONS.getFadvise()) { + gen.writeStringField("fadvise", value.getFadvise().name()); + } + if (value.isFastFailOnNotFoundEnabled() != DEFAULT_OPTIONS.isFastFailOnNotFoundEnabled()) { + gen.writeBooleanField("fastFailOnNotFoundEnabled", value.isFastFailOnNotFoundEnabled()); + } + if (value.getMinRangeRequestSize() != DEFAULT_OPTIONS.getMinRangeRequestSize()) { + gen.writeNumberField("minRangeRequestSize", value.getMinRangeRequestSize()); + } + if (value.getInplaceSeekLimit() != DEFAULT_OPTIONS.getInplaceSeekLimit()) { + gen.writeNumberField("inplaceSeekLimit", value.getInplaceSeekLimit()); + } + if (value.isGrpcReadZeroCopyEnabled() != DEFAULT_OPTIONS.isGrpcReadZeroCopyEnabled()) { + gen.writeBooleanField("grpcReadZeroCopyEnabled", value.isGrpcReadZeroCopyEnabled()); + } + gen.writeEndObject(); + } +} + +class GcsReadOptionsDeserializer extends JsonDeserializer { + @Override + public GoogleCloudStorageReadOptions deserialize(JsonParser p, DeserializationContext ctxt) + throws java.io.IOException { + JsonNode root = p.readValueAsTree(); + GoogleCloudStorageReadOptions.Builder builder = + GcsReadOptionsSerializer.DEFAULT_OPTIONS.toBuilder(); + + if (root != null && root.isObject()) { + if (root.hasNonNull("fadvise")) { + builder.setFadvise( + GoogleCloudStorageReadOptions.Fadvise.valueOf(root.get("fadvise").asText())); + } + if (root.hasNonNull("fastFailOnNotFoundEnabled")) { + builder.setFastFailOnNotFoundEnabled(root.get("fastFailOnNotFoundEnabled").asBoolean()); + } else if (root.hasNonNull("fastFailOnNotFound")) { + builder.setFastFailOnNotFoundEnabled(root.get("fastFailOnNotFound").asBoolean()); + } + if (root.hasNonNull("minRangeRequestSize")) { + builder.setMinRangeRequestSize(root.get("minRangeRequestSize").asLong()); + } + if (root.hasNonNull("inplaceSeekLimit")) { + builder.setInplaceSeekLimit(root.get("inplaceSeekLimit").asLong()); + } + if (root.hasNonNull("grpcReadZeroCopyEnabled")) { + builder.setGrpcReadZeroCopyEnabled(root.get("grpcReadZeroCopyEnabled").asBoolean()); + } + } + return builder.build(); + } +} 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 8af5e2260fc5..cd51d4fd9d28 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 void testGcpCoreApiSurface() throws Exception { final Set>> 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"), 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 c499290b851d..912f0110e9ce 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,46 @@ public void testEntriesWithErrors() throws Exception { 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.RANDOM) + .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.RANDOM, deserializedReadOptions.getFadvise()); + assertFalse(deserializedReadOptions.isFastFailOnNotFoundEnabled()); + assertEquals(12345L, deserializedReadOptions.getMinRangeRequestSize()); + } + + @Test + public void testDefaultGoogleCloudStorageReadOptionsSerialization() throws Exception { + GcsOptions options = PipelineOptionsFactory.as(GcsOptions.class); + 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()); + } }