Skip to content

Attempt to revert gcs connector to 2.2.26 #39698

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
shunping wants to merge 4 commits into
base: master
Choose a base branch
from
Draft

Attempt to revert gcs connector to 2.2.26 #39698

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2d96274
Revert "Fix gcs endpoint pipeline option wiring (#39435)"
shunping Aug 10, 2026
4e9e831
Revert "Suppress log spams in gcsio 3.0 (#38588)"
shunping Aug 10, 2026
0d04006
Revert "Upgrade gcsio to 3.1.16 (#38419)"
shunping Aug 10, 2026
f1c6de3
Revert "Override default fadvise to fix regression from gcs-connector…
shunping Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -623,7 +623,7 @@ class BeamModulePlugin implements Plugin<Project> {
def gax_version = "2.82.0"
def google_ads_version = "33.0.0"
def google_clients_version = "2.0.0"
def google_cloud_bigdataoss_version = "3.1.16"
def google_cloud_bigdataoss_version = "2.2.26"
def google_code_gson_version = "2.10.1"
def google_oauth_clients_version = "1.34.1"
// [bomupgrader] determined by: io.grpc:grpc-netty, consistent with: google_cloud_platform_libraries_bom
Expand Down Expand Up @@ -712,9 +712,9 @@ class BeamModulePlugin implements Plugin<Project> {
aws_java_sdk2_profiles : "software.amazon.awssdk:profiles:$aws_java_sdk2_version",
azure_sdk_bom : "com.azure:azure-sdk-bom:1.2.14",
bigdataoss_gcsio : "com.google.cloud.bigdataoss:gcsio:$google_cloud_bigdataoss_version",
bigdataoss_gcs_connector : "com.google.cloud.bigdataoss:gcs-connector:$google_cloud_bigdataoss_version",
bigdataoss_gcs_connector : "com.google.cloud.bigdataoss:gcs-connector:hadoop2-$google_cloud_bigdataoss_version",
bigdataoss_util : "com.google.cloud.bigdataoss:util:$google_cloud_bigdataoss_version",
bigdataoss_util_hadoop : "com.google.cloud.bigdataoss:util-hadoop:$google_cloud_bigdataoss_version",
bigdataoss_util_hadoop : "com.google.cloud.bigdataoss:util-hadoop:hadoop2-$google_cloud_bigdataoss_version",
byte_buddy : "net.bytebuddy:byte-buddy:1.17.7",
cassandra_driver_core : "com.datastax.cassandra:cassandra-driver-core:$cassandra_driver_version",
cassandra_driver_mapping : "com.datastax.cassandra:cassandra-driver-mapping:$cassandra_driver_version",
Expand Down
1 change: 0 additions & 1 deletion sdks/java/extensions/google-cloud-platform-core/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,6 @@ 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")
Expand Down
85 changes: 2 additions & 83 deletions ...ud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,6 @@
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;
Expand Down Expand Up @@ -58,16 +49,12 @@ public interface GcsOptions extends ApplicationNameOptions, GcpOptions, Pipeline
class GcsReadOptionsFactory implements DefaultValueFactory<GoogleCloudStorageReadOptions> {
@Override
public GoogleCloudStorageReadOptions create(PipelineOptions options) {
// 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;
return GoogleCloudStorageReadOptions.DEFAULT;
}
}

/** @deprecated This option will be removed in a future release. */
@JsonSerialize(using = GcsReadOptionsSerializer.class)
@JsonDeserialize(using = GcsReadOptionsDeserializer.class)
@JsonIgnore
@Description(
"The GoogleCloudStorageReadOptions instance that should be used to read from Google Cloud Storage.")
@Default.InstanceFactory(GcsReadOptionsFactory.class)
Expand Down Expand Up @@ -299,71 +286,3 @@ boolean exceedsEntryLimit() {
}
}
}

class GcsReadOptionsSerializer extends JsonSerializer<GoogleCloudStorageReadOptions> {
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<GoogleCloudStorageReadOptions> {
@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();
}
}
86 changes: 48 additions & 38 deletions ...-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import com.google.api.client.http.HttpHeaders;
import com.google.api.client.http.HttpRequestInitializer;
import com.google.api.client.http.HttpStatusCodes;
import com.google.api.client.http.HttpTransport;
import com.google.api.client.util.BackOff;
import com.google.api.client.util.Sleeper;
import com.google.api.services.storage.Storage;
Expand All @@ -52,6 +53,7 @@
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Method;
import java.nio.channels.SeekableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.nio.file.AccessDeniedException;
Expand All @@ -71,7 +73,6 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Supplier;
Expand Down Expand Up @@ -185,7 +186,6 @@ public boolean shouldRetry(IOException e) {
return RetryDeterminer.SOCKET_ERRORS.shouldRetry(e);
}
};
private static final AtomicBoolean overwriteLog = new AtomicBoolean(false);

/////////////////////////////////////////////////////////////////////////////

Expand Down Expand Up @@ -261,24 +261,14 @@ public boolean shouldRetry(IOException e) {
this.credentials = credentials;
this.maxBytesRewrittenPerCall = null;
this.numRewriteTokensUsed = null;
GoogleCloudStorageOptions.Builder optionsBuilder =
googleCloudStorageOptions =
GoogleCloudStorageOptions.builder()
.setAppName("Beam")
.setReadChannelOptions(gcsReadOptions)
.setGrpcEnabled(shouldUseGrpc);
if (storageClient.getRootUrl() != null) {
optionsBuilder.setStorageRootUrl(storageClient.getRootUrl());
}
if (storageClient.getServicePath() != null) {
optionsBuilder.setStorageServicePath(storageClient.getServicePath());
}
googleCloudStorageOptions = optionsBuilder.build();
try {
googleCloudStorage =
createGoogleCloudStorage(googleCloudStorageOptions, storageClient, credentials);
} catch (IOException e) {
throw new RuntimeException(e);
}
.setGrpcEnabled(shouldUseGrpc)
.build();
googleCloudStorage =
createGoogleCloudStorage(googleCloudStorageOptions, storageClient, credentials);
this.batchRequestSupplier =
() -> {
// Capture reference to this so that the most recent storageClient and initializer
Expand Down Expand Up @@ -501,11 +491,6 @@ private Long toFileSize(StorageObjectOrIOException storageObjectOrIOException)
}
}

@VisibleForTesting
GoogleCloudStorage getGoogleCloudStorage() {
return googleCloudStorage;
}

@VisibleForTesting
void setCloudStorageImpl(GoogleCloudStorage g) {
googleCloudStorage = g;
Expand Down Expand Up @@ -739,24 +724,49 @@ public WritableByteChannel create(GcsPath path, CreateOptions options) throws IO
}
}

@SuppressFBWarnings("LG_LOST_LOGGER_DUE_TO_WEAK_REFERENCE")
GoogleCloudStorage createGoogleCloudStorage(
GoogleCloudStorageOptions options, Storage storage, Credentials credentials)
throws IOException {
// Suppress log spams in gcsio 3.0
if (overwriteLog.compareAndSet(false, true)) {
java.util.logging.Logger.getLogger("com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl")
.setLevel(java.util.logging.Level.SEVERE);
GoogleCloudStorageOptions options, Storage storage, Credentials credentials) {
try {
return new GoogleCloudStorageImpl(options, storage, credentials);
} catch (NoSuchMethodError e) {
// gcs-connector 3.x drops the direct constructor and exclusively uses Builder
// TODO eliminate reflection once Beam drops Java 8 support and upgrades to gcsio 3.x
try {
final Method builderMethod = GoogleCloudStorageImpl.class.getMethod("builder");
Object builder = builderMethod.invoke(null);
final Class<?> builderClass =
Class.forName(
"com.google.cloud.hadoop.gcsio.AutoBuilder_GoogleCloudStorageImpl_Builder");

final Method setOptionsMethod =
builderClass.getMethod("setOptions", GoogleCloudStorageOptions.class);
setOptionsMethod.setAccessible(true);
builder = setOptionsMethod.invoke(builder, options);

final Method setHttpTransportMethod =
builderClass.getMethod("setHttpTransport", HttpTransport.class);
setHttpTransportMethod.setAccessible(true);
builder =
setHttpTransportMethod.invoke(builder, storage.getRequestFactory().getTransport());

final Method setCredentialsMethod =
builderClass.getMethod("setCredentials", Credentials.class);
setCredentialsMethod.setAccessible(true);
builder = setCredentialsMethod.invoke(builder, credentials);

final Method setHttpRequestInitializerMethod =
builderClass.getMethod("setHttpRequestInitializer", HttpRequestInitializer.class);
setHttpRequestInitializerMethod.setAccessible(true);
builder = setHttpRequestInitializerMethod.invoke(builder, httpRequestInitializer);

final Method buildMethod = builderClass.getMethod("build");
buildMethod.setAccessible(true);
return (GoogleCloudStorage) buildMethod.invoke(builder);
} catch (Exception reflectionError) {
throw new RuntimeException(
"Failed to construct GoogleCloudStorageImpl from gcsio 3.x Builder", reflectionError);
}
}

return GoogleCloudStorageImpl.builder()
.setOptions(options)
.setHttpTransport(storage.getRequestFactory().getTransport())
.setCredentials(credentials)
// gcsio 3 expects httpRequestInitializer to be either absent or
// com.google.cloud.hadoop.util.RetryHttpInitializer when credentials not provided
.setHttpRequestInitializer(credentials != null ? httpRequestInitializer : null)
.build();
}

/**
Expand Down
2 changes: 0 additions & 2 deletions ...platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/GcpCoreApiSurfaceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,6 @@ public void testGcpCoreApiSurface() throws Exception {
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"),
Expand Down
47 changes: 0 additions & 47 deletions ...latform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,16 +18,11 @@
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;
Expand Down Expand Up @@ -80,46 +75,4 @@ 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());
}
}
Loading
Loading