This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 97690fe9a9 GCS, add watchdog to monitor hanging transfer, fixes #8821
(#8822)
97690fe9a9 is described below
commit 97690fe9a91c8dff04c240f7beeb7b2791a16e00
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Sat Oct 10 10:14:20 2026 +0200
GCS, add watchdog to monitor hanging transfer, fixes #8821 (#8822)
* GCS, add watchdog to monitor hanging transfer, fixes #8821
* resolve remarks
---
.../ROOT/pages/vfs/google-cloud-storage-vfs.adoc | 33 ++
.../hop/vfs/gs/ComposeAppendOutputStream.java | 30 +-
.../apache/hop/vfs/gs/GoogleStorageFileObject.java | 35 +-
.../hop/vfs/gs/GoogleStorageFileProvider.java | 154 ++++--
.../apache/hop/vfs/gs/GoogleStorageFileSystem.java | 75 ++-
.../gs/GoogleStorageFileSystemConfigBuilder.java | 15 +
.../hop/vfs/gs/GoogleStorageObjectContext.java | 62 +++
.../hop/vfs/gs/GoogleStorageStallWatchdog.java | 513 ++++++++++++++++++++
.../apache/hop/vfs/gs/GoogleStorageVfsPlugin.java | 2 +-
.../hop/vfs/gs/LoggingStorageRetryStrategy.java | 196 ++++++++
.../apache/hop/vfs/gs/ReadChannelInputStream.java | 53 ++-
.../hop/vfs/gs/WriteChannelOutputStream.java | 89 +++-
.../hop/vfs/gs/config/GoogleCloudConfigPlugin.java | 28 +-
.../vfs/gs/config/GoogleCloudConfigSingleton.java | 2 +-
.../hop/vfs/gs/messages/messages_en_US.properties | 28 ++
.../hop/vfs/gs/ComposeAppendOutputStreamTest.java | 47 ++
.../hop/vfs/gs/GoogleStorageCredentialsTest.java | 161 +++++++
.../hop/vfs/gs/GoogleStorageRetryLoggingTest.java | 267 +++++++++++
.../hop/vfs/gs/GoogleStorageStallWatchdogTest.java | 516 +++++++++++++++++++++
.../vfs/gs/GoogleStorageStalledDownloadTest.java | 442 ++++++++++++++++++
.../hop/vfs/gs/ReadChannelInputStreamTest.java | 83 ++++
21 files changed, 2726 insertions(+), 105 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/vfs/google-cloud-storage-vfs.adoc
b/docs/hop-user-manual/modules/ROOT/pages/vfs/google-cloud-storage-vfs.adoc
index 6b6b0bed54..8e97bc9d06 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/vfs/google-cloud-storage-vfs.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/vfs/google-cloud-storage-vfs.adoc
@@ -53,12 +53,45 @@ Once done you will see a `googleCloud` entry in the central
`hop-config.json` fi
TIP: Running Hop in Docker? See
xref:docker-container.adoc#CloudCredentials[Cloud storage credentials] to pass
these settings to the container.
+Without a key file, Hop falls back to Application Default Credentials, for
example those of `gcloud auth application-default login` or of the machine it
runs on in Google Cloud.
+If no credentials can be found, Hop does not report it when it starts, as most
installations never use `gs://`.
+The first use of `gs://` fails instead, with an error that says why, for
example that the key file could not be read.
+
== Options
The options below are the ones in the *Plugins* section of the
xref:hop-gui/perspective-configuration.adoc[configuration perspective]. Each is
also settable with `hop-conf`.
include::partial$config/GoogleCloudStorageConfigPlugin.adoc[]
+== Retries, timeouts and stalled transfers
+
+The Google Cloud Storage client retries timeouts and temporary errors, such as
HTTP 503, by itself, following the options above.
+Hop logs every retry, so a slow or unreliable connection does not look like a
hang:
+
+[source]
+----
+Google Cloud Storage: No response in time while accessing gs://bucket/data.csv
(Read timed out), trying again. The connection timeout is 20 seconds, the read
timeout 20 seconds.
+----
+
+A download can also stall without failing.
+The client then reconnects and tries again within the same read, and never
gives up by itself, whatever the number of attempts.
+Hop therefore watches every read and write of a `gs://` file:
+
+* A read that makes no progress for three read timeouts, and at least 60
seconds, is reported.
+It is reported again after every further period without progress, and once
more if it starts moving again.
+* A read that still makes no progress after the time the retry options allow
for one request is ended.
+The transform reading the file then fails with an error that ends in `giving
up`.
+That time is *Max number of attempts* × (*Connect timeout* + *Read timeout*),
plus the retry delays between the attempts, and at most the *Total timeout*.
+With the default options that is 6 × (20 + 20) + 1 + 2 + 4 + 8 + 16 = 271
seconds.
+With an unlimited number of attempts (`0`), or a connect or read timeout of
`0` (no timeout), only the total timeout applies.
+When the total timeout is shorter than the time after which a read is
reported, the read is reported and ended when the total timeout is reached.
+To wait longer before giving up, raise the number of attempts or the timeouts.
+* An upload sends the file in parts of 16 MB, so a slow upload cannot be told
apart from a stalled one.
+A write is reported when one part takes longer than it would at 100 KB per
second, 164 seconds with the default options.
+A write is never ended.
+
+These messages are written to the general Hop log with the prefix `Google
Cloud Storage:`, for example in the console output of `hop-run`.
+
== Usage and testing
To test if the configuration works type use File/Open in the GUI and type in
gs:// as a file location.
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java
index 283630911a..8f72227a6c 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java
@@ -65,12 +65,34 @@ public class ComposeAppendOutputStream extends OutputStream
{
String tempName,
long targetGeneration,
WriteChannel tempChannel) {
+ this(
+ storage,
+ bucketName,
+ targetName,
+ tempName,
+ targetGeneration,
+ tempChannel,
+ GoogleStorageStallWatchdog.Transfer.untracked(null));
+ }
+
+ /**
+ * @param transfer the target object, named in the retry log and watched for
stalls while the
+ * appended bytes are uploaded
+ */
+ ComposeAppendOutputStream(
+ Storage storage,
+ String bucketName,
+ String targetName,
+ String tempName,
+ long targetGeneration,
+ WriteChannel tempChannel,
+ GoogleStorageStallWatchdog.Transfer transfer) {
this.storage = storage;
this.bucketName = bucketName;
this.targetName = targetName;
this.tempName = tempName;
this.targetGeneration = targetGeneration;
- this.tempStream = new WriteChannelOutputStream(tempChannel);
+ this.tempStream = new WriteChannelOutputStream(tempChannel, transfer);
}
@Override
@@ -115,7 +137,11 @@ public class ComposeAppendOutputStream extends
OutputStream {
storage.delete(BlobId.of(bucketName, tempName));
} catch (RuntimeException e) {
LogChannel.GENERAL.logError(
- "Unable to delete temporary append object gs://" + bucketName + "/"
+ tempName, e);
+ "Google Cloud Storage: Unable to delete temporary append object
gs://"
+ + bucketName
+ + "/"
+ + tempName,
+ e);
}
}
}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
index c43b8e61ae..08dec8316b 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
@@ -93,6 +93,12 @@ public class GoogleStorageFileObject extends
AbstractFileObject<GoogleStorageFil
@Override
protected void doAttach() throws Exception {
+ try (GoogleStorageObjectContext.Scope ignored =
GoogleStorageObjectContext.enter(uri())) {
+ attach();
+ }
+ }
+
+ private void attach() throws Exception {
Storage storage = getAbstractFileSystem().setupStorage();
if (!bucketName.isEmpty()) {
if (this.bucket == null) {
@@ -152,9 +158,24 @@ public class GoogleStorageFileObject extends
AbstractFileObject<GoogleStorageFil
}
Storage storage = getAbstractFileSystem().setupStorage();
if (blob != null) {
- return new ReadChannelInputStream(storage.reader(blob.getBlobId()));
+ return new ReadChannelInputStream(storage.reader(blob.getBlobId()),
reading());
}
- return new ReadChannelInputStream(storage.reader(BlobId.of(bucketName,
bucketPath)));
+ return new ReadChannelInputStream(storage.reader(BlobId.of(bucketName,
bucketPath)), reading());
+ }
+
+ /** How this object is named in the retry and stall log. */
+ private String uri() {
+ return getName().getFriendlyURI();
+ }
+
+ private GoogleStorageStallWatchdog.Transfer reading() throws IOException {
+ return GoogleStorageStallWatchdog.getInstance()
+ .reading(uri(), getAbstractFileSystem().stallLimits());
+ }
+
+ private GoogleStorageStallWatchdog.Transfer writing() throws IOException {
+ return GoogleStorageStallWatchdog.getInstance()
+ .writing(uri(), getAbstractFileSystem().stallLimits());
}
@Override
@@ -357,7 +378,7 @@ public class GoogleStorageFileObject extends
AbstractFileObject<GoogleStorageFil
this.blob = storage.create(BlobInfo.newBuilder(bucket,
objectName).build());
}
getAbstractFileSystem().invalidateListCacheForParentOf(bucketName,
bucketPath);
- return new WriteChannelOutputStream(storage.writer(blob));
+ return new WriteChannelOutputStream(storage.writer(blob), writing());
}
/**
@@ -366,12 +387,14 @@ public class GoogleStorageFileObject extends
AbstractFileObject<GoogleStorageFil
* target. See <a
href="https://cloud.google.com/storage/docs/composite-objects#appends";>GCS
* appends</a>.
*/
- private OutputStream openComposeAppendStream(Storage storage, Blob current,
String objectName) {
+ private OutputStream openComposeAppendStream(Storage storage, Blob current,
String objectName)
+ throws IOException {
String tempName = objectName + ".hop-append-" + UUID.randomUUID() + ".tmp";
+ GoogleStorageStallWatchdog.Transfer transfer = writing();
WriteChannel tempChannel = storage.writer(BlobInfo.newBuilder(bucketName,
tempName).build());
getAbstractFileSystem().invalidateListCacheForParentOf(bucketName,
bucketPath);
return new ComposeAppendOutputStream(
- storage, bucketName, objectName, tempName, current.getGeneration(),
tempChannel);
+ storage, bucketName, objectName, tempName, current.getGeneration(),
tempChannel, transfer);
}
@Override
@@ -482,7 +505,7 @@ public class GoogleStorageFileObject extends
AbstractFileObject<GoogleStorageFil
return Objects.hash(getName().getPath());
}
- private OffsetDateTime getLatestModifiedFileTime() {
+ private OffsetDateTime getLatestModifiedFileTime() throws IOException {
Storage storage = getAbstractFileSystem().setupStorage();
OffsetDateTime latest = OffsetDateTime.ofInstant(Instant.EPOCH,
ZoneOffset.UTC);
Page<Blob> page =
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
index 5ccde11bf5..1cfd95a8ba 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
@@ -21,6 +21,8 @@ package org.apache.hop.vfs.gs;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.auth.oauth2.ServiceAccountCredentials;
import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Set;
@@ -41,6 +43,9 @@ import
org.apache.hop.vfs.gs.metadatatype.GoogleStorageMetadataType;
public class GoogleStorageFileProvider extends AbstractOriginatingFileProvider
{
private FileSystemOptions newFileSystemOptions = new FileSystemOptions();
+ /** Why {@link #tryApplicationDefaultCredentials()} last came back
empty-handed. */
+ private String applicationDefaultProblem;
+
public GoogleStorageFileProvider() {
super();
setServiceAccountCredentials(null, null);
@@ -80,32 +85,59 @@ public class GoogleStorageFileProvider extends
AbstractOriginatingFileProvider {
return new GoogleStorageFileSystem(rootName, null, newFileSystemOptions);
}
+ /**
+ * Load the credentials for this provider. A problem is not logged here for
the default {@code
+ * gs://} scheme: every Hop installation creates that provider, also where
nobody uses Google
+ * Cloud Storage. Instead the reason is kept with the file system options,
so the first actual use
+ * of {@code gs://} can report it and fail clearly, see {@link
+ * GoogleStorageFileSystem#setupStorage()}.
+ */
private void setServiceAccountCredentials(
IVariables variables, GoogleStorageMetadataType
googleStorageMetadataType) {
+ GoogleStorageFileSystemConfigBuilder builder =
+ GoogleStorageFileSystemConfigBuilder.getInstance();
+ boolean defaultScheme = variables == null && googleStorageMetadataType ==
null;
+ String scheme = defaultScheme ? "gs" : googleStorageMetadataType.getName();
+ builder.setSchema(newFileSystemOptions, scheme);
try {
- GoogleCredentials credentials = null;
- String scheme = "gs";
-
- if (variables == null && googleStorageMetadataType == null) {
+ GoogleCredentials credentials;
+ if (defaultScheme) {
// Default configuration: prefer explicit key file; ADC only as
optional fallback.
// Do not call ADC first — on non-GCP clusters (e.g. Databricks AWS)
it probes metadata
// and can throw NoClassDefFoundError for io.grpc.Context when gRPC is
not on the fat jar.
GoogleCloudConfig config = GoogleCloudConfigSingleton.getConfig();
- if (!StringUtils.isEmpty(config.getServiceAccountKeyFile())) {
- credentials =
- ServiceAccountCredentials.fromStream(
- new FileInputStream(config.getServiceAccountKeyFile()));
+ String keyFile = config.getServiceAccountKeyFile();
+ if (!StringUtils.isEmpty(keyFile)) {
+ try {
+ credentials = readKeyFile(keyFile);
+ } catch (Exception | LinkageError e) {
+ // A LinkageError too: the handler below is about Application
Default Credentials.
+ builder.setCredentialsProblem(
+ newFileSystemOptions,
+ "the service account key file '"
+ + keyFile
+ + "' set in the Google Cloud options could not be read ("
+ + describe(e)
+ + ")");
+ return;
+ }
} else {
credentials = tryApplicationDefaultCredentials();
+ if (credentials == null) {
+ builder.setCredentialsProblem(
+ newFileSystemOptions,
+ "no service account key file is set in the Google Cloud
options, and Application"
+ + " Default Credentials are not available ("
+ + applicationDefaultProblem
+ + "). Set a key file, or run 'gcloud auth
application-default login'");
+ return;
+ }
}
} else {
- scheme = googleStorageMetadataType.getName();
switch (googleStorageMetadataType.getStorageCredentialsType()) {
case KEY_FILE:
credentials =
- ServiceAccountCredentials.fromStream(
- new FileInputStream(
-
variables.resolve(googleStorageMetadataType.getStorageAccountKey())));
+
readKeyFile(variables.resolve(googleStorageMetadataType.getStorageAccountKey()));
break;
case KEY_STRING:
credentials =
@@ -118,60 +150,94 @@ public class GoogleStorageFileProvider extends
AbstractOriginatingFileProvider {
credentials = tryApplicationDefaultCredentials();
break;
}
+ if (credentials == null) {
+ builder.setCredentialsProblem(
+ newFileSystemOptions,
+ "Google Storage connection '"
+ + scheme
+ + "' uses Application Default Credentials, which are not
available ("
+ + applicationDefaultProblem
+ + ")");
+ return;
+ }
}
- if (credentials != null) {
- GoogleStorageFileSystemConfigBuilder.getInstance()
- .setGoogleCredentials(newFileSystemOptions, credentials);
- }
-
GoogleStorageFileSystemConfigBuilder.getInstance().setSchema(newFileSystemOptions,
scheme);
+ builder.setGoogleCredentials(newFileSystemOptions, credentials);
} catch (Exception e) {
- // Do not log error for the default GS account
- if (googleStorageMetadataType != null) {
- LogChannel.GENERAL.logError(
- "Unable to set service account credentials for vfs name: "
- + googleStorageMetadataType.getName(),
- e);
- }
+ // Only a named connection gets here: it is configured on purpose, so
say so right away.
+ LogChannel.GENERAL.logError(
+ "Google Cloud Storage: Unable to set service account credentials for
vfs name: " + scheme,
+ e);
+ builder.setCredentialsProblem(
+ newFileSystemOptions,
+ "the credentials of Google Storage connection '"
+ + scheme
+ + "' could not be loaded ("
+ + describe(e)
+ + ")");
} catch (LinkageError e) {
// NoClassDefFoundError (e.g. io.grpc.Context) must not prevent
HopEnvironment.init
- if (googleStorageMetadataType != null) {
+ if (defaultScheme) {
+ LogChannel.GENERAL.logDetailed(
+ "Google Cloud Storage: Application Default Credentials unavailable
("
+ + describe(e)
+ + "). gs:// will work after configuring a service account
key.");
+ builder.setCredentialsProblem(
+ newFileSystemOptions,
+ "Application Default Credentials are not available ("
+ + describe(e)
+ + "). Set a service account key file in the Google Cloud
options");
+ } else {
LogChannel.GENERAL.logError(
- "Unable to set service account credentials for vfs name: "
- + googleStorageMetadataType.getName()
+ "Google Cloud Storage: Unable to set service account credentials
for vfs name: "
+ + scheme
+ " (missing dependency: "
+ e.getMessage()
+ ")",
e);
- } else {
- LogChannel.GENERAL.logDetailed(
- "Google Storage VFS: Application Default Credentials unavailable ("
- + e.getClass().getSimpleName()
- + ": "
+ builder.setCredentialsProblem(
+ newFileSystemOptions,
+ "the credentials of Google Storage connection '"
+ + scheme
+ + "' could not be loaded (missing dependency: "
+ e.getMessage()
- + "). gs:// will work after configuring a service account
key.");
+ + ")");
}
}
}
+ /**
+ * A plain file, not Hop VFS: this runs while the VFS providers themselves
are being set up.
+ *
+ * @param keyFile the service account key file
+ * @return the credentials in it
+ */
+ private static GoogleCredentials readKeyFile(String keyFile) throws
IOException {
+ try (InputStream in = new FileInputStream(keyFile)) {
+ return ServiceAccountCredentials.fromStream(in);
+ }
+ }
+
/**
* Best-effort ADC. Returns null if unavailable (not on GCE/GCP, or
gRPC/auth deps missing from
- * classpath — common with native-provided fat jars on Databricks/AWS).
+ * classpath — common with native-provided fat jars on Databricks/AWS),
keeping the reason in
+ * {@link #applicationDefaultProblem}.
*/
- static GoogleCredentials tryApplicationDefaultCredentials() {
+ GoogleCredentials tryApplicationDefaultCredentials() {
try {
return GoogleCredentials.getApplicationDefault();
- } catch (Exception e) {
+ } catch (Exception | LinkageError e) {
+ applicationDefaultProblem = describe(e);
LogChannel.GENERAL.logDetailed(
- "Google Storage VFS: Application Default Credentials not available:
" + e.getMessage());
- return null;
- } catch (LinkageError e) {
- LogChannel.GENERAL.logDetailed(
- "Google Storage VFS: Application Default Credentials not available ("
- + e.getClass().getSimpleName()
- + ": "
- + e.getMessage()
+ "Google Cloud Storage: Application Default Credentials not available
("
+ + applicationDefaultProblem
+ ")");
return null;
}
}
+
+ private static String describe(Throwable e) {
+ return e.getMessage() == null
+ ? e.getClass().getSimpleName()
+ : e.getClass().getSimpleName() + ": " + e.getMessage();
+ }
}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystem.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystem.java
index 0d6130a2c0..aa6594359c 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystem.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystem.java
@@ -20,6 +20,7 @@ package org.apache.hop.vfs.gs;
import com.google.api.gax.core.FixedCredentialsProvider;
import com.google.api.gax.retrying.RetrySettings;
+import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.http.HttpTransportOptions;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
@@ -28,6 +29,7 @@ import com.google.storage.control.v2.StorageControlClient;
import com.google.storage.control.v2.StorageControlSettings;
import java.io.IOException;
import java.util.Collection;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.vfs2.Capability;
import org.apache.commons.vfs2.FileName;
import org.apache.commons.vfs2.FileObject;
@@ -36,18 +38,29 @@ import org.apache.commons.vfs2.FileSystemOptions;
import org.apache.commons.vfs2.provider.AbstractFileName;
import org.apache.commons.vfs2.provider.AbstractFileSystem;
import org.apache.hop.core.Const;
+import org.apache.hop.core.logging.LogChannel;
import org.apache.hop.vfs.gs.config.GoogleCloudConfig;
import org.apache.hop.vfs.gs.config.GoogleCloudConfigSingleton;
import org.threeten.bp.Duration;
public class GoogleStorageFileSystem extends AbstractFileSystem {
- Storage storage = null;
+ /**
+ * Volatile, and set after {@link #stallLimits}: a thread that finds the
client built by another
+ * thread must also find the limits that go with it.
+ */
+ volatile Storage storage = null;
+
+ /** When to report or end a transfer, from the same settings {@link
#storage} was built with. */
+ private GoogleStorageStallWatchdog.Limits stallLimits;
+
StorageControlClient storageControlClient = null;
FileSystemOptions fileSystemOptions;
private GoogleStorageListCache listCache;
+ private final AtomicBoolean credentialsProblemLogged = new AtomicBoolean();
+
protected GoogleStorageFileSystem(
FileName rootName, FileObject parentLayer, FileSystemOptions
fileSystemOptions)
throws FileSystemException {
@@ -77,7 +90,7 @@ public class GoogleStorageFileSystem extends
AbstractFileSystem {
caps.addAll(GoogleStorageFileProvider.capabilities);
}
- Storage setupStorage() {
+ Storage setupStorage() throws IOException {
if (storage != null) {
return storage;
}
@@ -85,9 +98,9 @@ public class GoogleStorageFileSystem extends
AbstractFileSystem {
GoogleCloudConfig config = GoogleCloudConfigSingleton.getConfig();
StorageOptions.Builder optionsBuilder = buildStorageOptions(config);
- optionsBuilder.setCredentials(
-
GoogleStorageFileSystemConfigBuilder.getInstance().getGoogleCredentials(fileSystemOptions));
+ optionsBuilder.setCredentials(credentials());
+ stallLimits = GoogleStorageStallWatchdog.Limits.from(config);
return storage = optionsBuilder.build().getService();
}
@@ -97,10 +110,23 @@ public class GoogleStorageFileSystem extends
AbstractFileSystem {
* wiring can be exercised from a test against a local endpoint.
*/
static StorageOptions.Builder buildStorageOptions(GoogleCloudConfig config) {
+ return buildStorageOptions(
+ config, new LoggingStorageRetryStrategy(selectRetryStrategy(config),
config));
+ }
+
+ /** As {@link #buildStorageOptions(GoogleCloudConfig)}, reporting retries to
the given log. */
+ static StorageOptions.Builder buildStorageOptions(
+ GoogleCloudConfig config, LoggingStorageRetryStrategy.RetryLog retryLog)
{
+ return buildStorageOptions(
+ config, new LoggingStorageRetryStrategy(selectRetryStrategy(config),
config, retryLog));
+ }
+
+ private static StorageOptions.Builder buildStorageOptions(
+ GoogleCloudConfig config, StorageRetryStrategy retryStrategy) {
return StorageOptions.newBuilder()
.setRetrySettings(buildRetrySettings(config))
.setTransportOptions(buildTransportOptions(config))
- .setStorageRetryStrategy(selectRetryStrategy(config));
+ .setStorageRetryStrategy(retryStrategy);
}
/**
@@ -137,6 +163,17 @@ public class GoogleStorageFileSystem extends
AbstractFileSystem {
.build();
}
+ /**
+ * The limits are taken from the settings the client was built with, so that
the watchdog and the
+ * client it watches always agree on the timeouts, even after the settings
changed.
+ *
+ * @return how long a read or write may go without progress before it is
reported or ended
+ */
+ GoogleStorageStallWatchdog.Limits stallLimits() throws IOException {
+ setupStorage();
+ return stallLimits;
+ }
+
static HttpTransportOptions buildTransportOptions(GoogleCloudConfig config) {
int connectTimeoutMs = Const.toInt(config.getConnectionTimeout(), 20) *
1000;
int readTimeoutMs = Const.toInt(config.getReadTimeout(), 20) * 1000;
@@ -186,6 +223,29 @@ public class GoogleStorageFileSystem extends
AbstractFileSystem {
}
}
+ /**
+ * The credentials the provider loaded. When it could not, the Google client
would fail with a
+ * bare NullPointerException; instead say why, once in the log and every
time in the error.
+ */
+ private GoogleCredentials credentials() throws IOException {
+ GoogleStorageFileSystemConfigBuilder builder =
+ GoogleStorageFileSystemConfigBuilder.getInstance();
+ GoogleCredentials credentials =
builder.getGoogleCredentials(fileSystemOptions);
+ if (credentials != null) {
+ return credentials;
+ }
+ String problem = builder.getCredentialsProblem(fileSystemOptions);
+ String message =
+ "Google Cloud Storage: no credentials to access "
+ + Const.NVL(builder.getSchema(fileSystemOptions), "gs")
+ + "://: "
+ + Const.NVL(problem, "none were configured");
+ if (!credentialsProblemLogged.getAndSet(true)) {
+ LogChannel.GENERAL.logError(message);
+ }
+ throw new IOException(message);
+ }
+
StorageControlClient getStorageControlClient() throws IOException {
if (storageControlClient != null) {
return storageControlClient;
@@ -193,10 +253,7 @@ public class GoogleStorageFileSystem extends
AbstractFileSystem {
RetrySettings retrySettings =
buildRetrySettings(GoogleCloudConfigSingleton.getConfig());
StorageControlSettings.Builder builder =
StorageControlSettings.newBuilder()
- .setCredentialsProvider(
- FixedCredentialsProvider.create(
- GoogleStorageFileSystemConfigBuilder.getInstance()
- .getGoogleCredentials(fileSystemOptions)));
+
.setCredentialsProvider(FixedCredentialsProvider.create(credentials()));
// This client was left on the library defaults, so the configured retry
behaviour never
// reached HNS folder operations.
builder.applyToAllUnaryMethods(
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystemConfigBuilder.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystemConfigBuilder.java
index 4909ae5782..ba13c24390 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystemConfigBuilder.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileSystemConfigBuilder.java
@@ -26,6 +26,7 @@ import org.apache.commons.vfs2.FileSystemOptions;
public class GoogleStorageFileSystemConfigBuilder extends
FileSystemConfigBuilder {
private static final String GOOGLE_CREDENTIALS = "googleCredentials";
+ private static final String CREDENTIALS_PROBLEM = "credentialsProblem";
private static final GoogleStorageFileSystemConfigBuilder builder =
new GoogleStorageFileSystemConfigBuilder();
@@ -43,6 +44,20 @@ public class GoogleStorageFileSystemConfigBuilder extends
FileSystemConfigBuilde
return (GoogleCredentials) getParam(opts, GOOGLE_CREDENTIALS);
}
+ /**
+ * Why no credentials could be loaded, reported on first use of the file
system.
+ *
+ * @param opts the file system options
+ * @param problem what went wrong, and what to do about it
+ */
+ public void setCredentialsProblem(FileSystemOptions opts, String problem) {
+ setParam(opts, CREDENTIALS_PROBLEM, problem);
+ }
+
+ public String getCredentialsProblem(FileSystemOptions opts) {
+ return (String) getParam(opts, CREDENTIALS_PROBLEM);
+ }
+
public void setSchema(FileSystemOptions opts, String schema) {
setParam(opts, "Schema", schema);
}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageObjectContext.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageObjectContext.java
new file mode 100644
index 0000000000..78d675149d
--- /dev/null
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageObjectContext.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.gs;
+
+/**
+ * The {@code gs://} object the current thread is working on, so that a retry
reported by {@link
+ * LoggingStorageRetryStrategy} can say which file it concerns. The storage
client retries on the
+ * calling thread, backoff included, which is what makes a thread-local the
right reach.
+ */
+final class GoogleStorageObjectContext {
+
+ private static final ThreadLocal<String> CURRENT = new ThreadLocal<>();
+
+ /** Restores the object the thread was working on before. */
+ interface Scope extends AutoCloseable {
+ @Override
+ void close();
+ }
+
+ private GoogleStorageObjectContext() {}
+
+ /**
+ * Mark the current thread as working on the given object until the returned
scope is closed.
+ *
+ * @param uri the object, for example {@code gs://bucket/folder/file.txt};
null when unknown
+ * @return the scope to close when the work is done
+ */
+ static Scope enter(String uri) {
+ String previous = CURRENT.get();
+ CURRENT.set(uri);
+ return () -> {
+ if (previous == null) {
+ CURRENT.remove();
+ } else {
+ CURRENT.set(previous);
+ }
+ };
+ }
+
+ /**
+ * @return the object the current thread is working on, or null when unknown
+ */
+ static String current() {
+ return CURRENT.get();
+ }
+}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdog.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdog.java
new file mode 100644
index 0000000000..51e7718035
--- /dev/null
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdog.java
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.gs;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import org.apache.hop.core.Const;
+import org.apache.hop.core.logging.LogChannel;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.vfs.gs.config.GoogleCloudConfig;
+
+/**
+ * Reports a {@code gs://} read or write that stopped making progress, and
ends a read that stays
+ * stuck.
+ *
+ * <p>The storage client can keep a single read call busy indefinitely: it
re-opens a stalled
+ * download over and over inside one call, ignoring the configured attempts
and total timeout, so
+ * the code calling it learns nothing. Retries are logged by {@link
LoggingStorageRetryStrategy},
+ * but a transfer that does not move is only visible from outside the call.
Each read or write marks
+ * itself busy while it is inside the client, and a single background thread
looks at those every so
+ * often: a call that has been busy for longer than the warning period is
logged, again after every
+ * further period, and once more if it gets going again.
+ *
+ * <p>A read that is still stuck after the time the retry settings allow is
interrupted. The client
+ * does not retry an interrupted read, so the call ends and the stream fails
with a clear error.
+ * Writes are never interrupted: the client uploads a whole 16 MB chunk inside
one call, so a slow
+ * upload cannot be told apart from a stuck one, and is only reported.
+ */
+final class GoogleStorageStallWatchdog {
+
+ private static final Class<?> PKG = GoogleStorageStallWatchdog.class;
+
+ private static final GoogleStorageStallWatchdog INSTANCE =
+ new GoogleStorageStallWatchdog(
+ message -> LogChannel.GENERAL.logBasic("Google Cloud Storage: " +
message),
+ System::nanoTime,
+ Duration.ofSeconds(1),
+ daemonInterrupter());
+
+ /** Where a stall is reported. */
+ @FunctionalInterface
+ interface StallLog {
+ void report(String message);
+ }
+
+ /** Whether a transfer reads or writes, which is all the messages need to
know. */
+ enum Direction {
+ READING("Reading"),
+ WRITING("Writing");
+
+ private final String key;
+
+ Direction(String key) {
+ this.key = key;
+ }
+ }
+
+ /**
+ * How long a call may go without progress before it is reported or, for a
read, ended.
+ *
+ * @param readWarning a read busy for this long is reported as stalled
+ * @param readAbort a read busy for this long is ended; zero for never
+ * @param writeWarning a write busy for this long is reported as slow or
stalled
+ */
+ record Limits(Duration readWarning, Duration readAbort, Duration
writeWarning) {
+
+ /** Never report a stall sooner than this, whatever the read timeout. */
+ private static final long MINIMUM_WARNING_SECONDS = 60;
+
+ /** The client's upload chunk, sent whole inside one write call. Hop
leaves it as it is. */
+ private static final long UPLOAD_CHUNK_BYTES = 16L * 1024 * 1024;
+
+ /** Below this an upload is too slow to tell from a stuck one. */
+ private static final long SLOWEST_UPLOAD_BYTES_PER_SECOND = 100L * 1024;
+
+ /**
+ * Derive the limits from the Google Cloud configuration.
+ *
+ * <ul>
+ * <li>A read is reported after three read timeouts without progress, at
least a minute: a
+ * single timed-out attempt is already reported as a retry.
+ * <li>A read is ended after the time the retry settings allow for one
call - every attempt
+ * running into both the connection and the read timeout, plus the
delays between them -
+ * and never before it was reported. Until then the client may still
give up by itself,
+ * with its own error. A timeout of zero is no timeout at all, so an
attempt is not
+ * bounded and neither is that time.
+ * <li>The total timeout caps both: a read is never ended later than
that, and when that comes
+ * before the warning, it is reported and ended at the same moment.
+ * <li>A write is reported once one upload chunk has taken longer than
it would at the slowest
+ * rate we still call working, and not before a read would be.
+ * </ul>
+ *
+ * @param config the Google Cloud configuration
+ * @return the limits
+ */
+ static Limits from(GoogleCloudConfig config) {
+ long readTimeout = timeoutSeconds(config.getReadTimeout());
+ long connectTimeout = timeoutSeconds(config.getConnectionTimeout());
+ Duration readWarning =
Duration.ofSeconds(Math.max(MINIMUM_WARNING_SECONDS, 3 * readTimeout));
+
+ Duration readAbort =
+ readTimeout == 0 || connectTimeout == 0
+ ? Duration.ZERO
+ : retryBudget(config, connectTimeout + readTimeout);
+ if (!readAbort.isZero() && readAbort.compareTo(readWarning) < 0) {
+ readAbort = readWarning;
+ }
+ long totalTimeoutMinutes = Const.toLong(config.getTotalTimeout(), 50);
+ if (totalTimeoutMinutes > 0) {
+ Duration totalTimeout = Duration.ofMinutes(totalTimeoutMinutes);
+ if (readAbort.isZero() || readAbort.compareTo(totalTimeout) > 0) {
+ readAbort = totalTimeout;
+ }
+ if (readWarning.compareTo(readAbort) > 0) {
+ readWarning = readAbort;
+ }
+ }
+
+ long chunkSeconds =
+ (UPLOAD_CHUNK_BYTES + SLOWEST_UPLOAD_BYTES_PER_SECOND - 1)
+ / SLOWEST_UPLOAD_BYTES_PER_SECOND;
+ Duration writeWarning =
Duration.ofSeconds(Math.max(readWarning.toSeconds(), chunkSeconds));
+
+ return new Limits(readWarning, readAbort, writeWarning);
+ }
+
+ /**
+ * A configured timeout in seconds, the way the client takes it: none at
all, or a negative one,
+ * is the 20 second default, and zero is no timeout.
+ */
+ private static long timeoutSeconds(String configured) {
+ long seconds = Const.toInt(configured, 20);
+ return seconds < 0 ? 20 : seconds;
+ }
+
+ /** Every attempt running into its timeouts, plus the delays between them.
*/
+ private static Duration retryBudget(GoogleCloudConfig config, long
attemptSeconds) {
+ int attempts = Const.toInt(config.getMaxAttempts(), 6);
+ if (attempts <= 0) {
+ // Unlimited attempts: only the total timeout bounds them.
+ return Duration.ZERO;
+ }
+ double delay = Const.toLong(config.getInitialRetryDelay(), 1);
+ double multiplier = Const.toDouble(config.getRetryDelayMultiplier(),
2.0);
+ long maxDelay = Const.toLong(config.getMaxRetryDelay(), 32);
+ double delays = 0;
+ for (int i = 1; i < attempts; i++) {
+ delays += Math.min(delay, maxDelay);
+ delay *= multiplier;
+ }
+ return Duration.ofSeconds(attempts * attemptSeconds + (long)
Math.ceil(delays));
+ }
+ }
+
+ private final StallLog log;
+ private final LongSupplier clock;
+ private final Duration checkInterval;
+ private final Executor interrupter;
+ private final Set<Transfer> busy = ConcurrentHashMap.newKeySet();
+ private ScheduledExecutorService checker;
+
+ /**
+ * @param log where stalls are reported
+ * @param clock the time in nanoseconds
+ * @param checkInterval how often to look for stalls; null to only look when
{@link #check()} is
+ * called
+ * @param interrupter runs the interrupts that end stuck reads
+ */
+ GoogleStorageStallWatchdog(
+ StallLog log, LongSupplier clock, Duration checkInterval, Executor
interrupter) {
+ this.log = log;
+ this.clock = clock;
+ this.checkInterval = checkInterval;
+ this.interrupter = interrupter;
+ }
+
+ /**
+ * Interrupting a stuck read makes the client close its HTTP stream, and
that close can wait for
+ * the blocked socket read to time out. That wait must not hold up the
checks on every other
+ * transfer, so interrupts get threads of their own.
+ *
+ * @return an executor running each interrupt on a daemon thread
+ */
+ static Executor daemonInterrupter() {
+ return Executors.newCachedThreadPool(
+ r -> {
+ Thread thread = new Thread(r, "Google Cloud Storage stall watchdog
interrupt");
+ thread.setDaemon(true);
+ return thread;
+ });
+ }
+
+ static GoogleStorageStallWatchdog getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * @param uri the object being read
+ * @param limits when to report the read, and when to end it
+ * @return a transfer to mark the reads with
+ */
+ Transfer reading(String uri, Limits limits) {
+ return new Transfer(this, Direction.READING, uri, limits.readWarning(),
limits.readAbort());
+ }
+
+ /**
+ * @param uri the object being written
+ * @param limits when to report the write
+ * @return a transfer to mark the writes with
+ */
+ Transfer writing(String uri, Limits limits) {
+ return new Transfer(this, Direction.WRITING, uri, limits.writeWarning(),
Duration.ZERO);
+ }
+
+ /** Report every busy transfer that has gone without progress for too long,
and end reads. */
+ void check() {
+ long now = clock.getAsLong();
+ for (Transfer transfer : busy) {
+ transfer.checkAt(now);
+ }
+ }
+
+ /** Stop looking for stalls. Only the tests have a reason to. */
+ synchronized void shutdown() {
+ if (checker != null) {
+ checker.shutdownNow();
+ checker = null;
+ }
+ }
+
+ private synchronized void startChecking() {
+ if (checker != null || checkInterval == null) {
+ return;
+ }
+ checker =
+ Executors.newSingleThreadScheduledExecutor(
+ r -> {
+ Thread thread = new Thread(r, "Google Cloud Storage stall
watchdog");
+ thread.setDaemon(true);
+ return thread;
+ });
+ long millis = checkInterval.toMillis();
+ checker.scheduleWithFixedDelay(this::checkQuietly, millis, millis,
TimeUnit.MILLISECONDS);
+ }
+
+ /** An exception escaping a scheduled task silently cancels it, so none may.
*/
+ private void checkQuietly() {
+ try {
+ check();
+ } catch (RuntimeException e) {
+ // Keep watching; the next round may well succeed.
+ }
+ }
+
+ private void report(String message) {
+ try {
+ log.report(message);
+ } catch (RuntimeException e) {
+ // Failing to log a stall must not fail the transfer or stop the
watchdog.
+ }
+ }
+
+ /**
+ * One object being read or written. The stream marks each call into the
client with {@link
+ * #begin()}, and with {@link #end(long)} or {@link #failed()} from a {@code
finally}. Calls may
+ * overlap - a stream's {@code close()} sends the last upload chunk while a
{@code write()} may
+ * still be inside the client - and the transfer stays busy until the last
of them is done.
+ */
+ static final class Transfer {
+ private final GoogleStorageStallWatchdog watchdog;
+ private final Direction direction;
+ private final String uri;
+ private final long warnAfterNanos;
+ private final long abortAfterNanos;
+
+ /**
+ * Held while an interrupt is delivered and while {@link #end(long)} lets
go of the thread, so
+ * that an interrupt never reaches a thread that has moved on. Taken
before the transfer's own
+ * lock, and never by the checks: delivering an interrupt can take as long
as a read timeout.
+ */
+ private final Object interruptLock = new Object();
+
+ private long bytes;
+
+ /** The threads inside the client right now, one entry per call. Busy
while not empty. */
+ private final List<Thread> threads = new ArrayList<>();
+
+ private long busySince;
+ private long nextReport;
+ private boolean stalled;
+ private String abortMessage;
+ private boolean interrupting;
+
+ private Transfer(
+ GoogleStorageStallWatchdog watchdog,
+ Direction direction,
+ String uri,
+ Duration warnAfter,
+ Duration abortAfter) {
+ this.watchdog = watchdog;
+ this.direction = direction;
+ this.uri = uri;
+ this.warnAfterNanos = warnAfter == null ? 0 : warnAfter.toNanos();
+ this.abortAfterNanos = abortAfter == null ? 0 : abortAfter.toNanos();
+ }
+
+ /**
+ * A transfer that is never reported or ended, for a stream that does not
know what it moves.
+ *
+ * @param uri the object, named in the retry log; may be null
+ * @return a transfer that does nothing
+ */
+ static Transfer untracked(String uri) {
+ return new Transfer(null, Direction.READING, uri, null, null);
+ }
+
+ /**
+ * @return the object being transferred, or null when unknown
+ */
+ String uri() {
+ return uri;
+ }
+
+ /**
+ * A call into the client is about to start on the current thread. While
another call is already
+ * inside the client, the time without progress keeps counting from when
that one began.
+ */
+ void begin() {
+ if (watchdog == null) {
+ return;
+ }
+ synchronized (this) {
+ if (threads.isEmpty()) {
+ busySince = watchdog.clock.getAsLong();
+ nextReport = busySince + warnAfterNanos;
+ stalled = false;
+ abortMessage = null;
+ interrupting = false;
+ watchdog.busy.add(this);
+ }
+ threads.add(Thread.currentThread());
+ }
+ watchdog.startChecking();
+ }
+
+ /**
+ * The call into the client on the current thread returned. When the
watchdog ended it, the
+ * interrupt it used is cleared here, so it does not leak into whatever
the thread does next.
+ *
+ * @param transferred the number of bytes it moved
+ * @return true when the watchdog ended the call; {@link
#abortedError(Throwable)} says why
+ */
+ boolean end(long transferred) {
+ return leave(transferred, true);
+ }
+
+ /**
+ * The call into the client on the current thread threw. A transfer that
was reported as stalled
+ * is not reported as resumed: the error says what happened.
+ *
+ * @return true when the watchdog ended the call; {@link
#abortedError(Throwable)} says why
+ */
+ boolean failed() {
+ return leave(0, false);
+ }
+
+ private boolean leave(long transferred, boolean returned) {
+ if (watchdog == null) {
+ return false;
+ }
+ String resumed = null;
+ boolean aborted;
+ synchronized (interruptLock) {
+ synchronized (this) {
+ bytes += Math.max(0, transferred);
+ threads.remove(Thread.currentThread());
+ aborted = abortMessage != null;
+ if (aborted) {
+ Thread.interrupted();
+ }
+ if (threads.isEmpty()) {
+ watchdog.busy.remove(this);
+ if (!aborted && stalled && returned) {
+ resumed =
+ BaseMessages.getString(
+ PKG,
+ "GoogleStorageStallWatchdog.Resumed." + direction.key,
+ uri,
+ seconds(watchdog.clock.getAsLong() - busySince));
+ }
+ stalled = false;
+ }
+ }
+ }
+ if (resumed != null) {
+ watchdog.report(resumed);
+ }
+ return aborted;
+ }
+
+ /**
+ * @param cause what the client threw when it was interrupted
+ * @return the error to fail the stream with after the watchdog ended a
call
+ */
+ synchronized IOException abortedError(Throwable cause) {
+ return new IOException(abortMessage, cause);
+ }
+
+ private void checkAt(long now) {
+ String message = null;
+ boolean interrupt = false;
+ synchronized (this) {
+ // The watchdog may still be holding this transfer from before it was
removed.
+ if (threads.isEmpty()) {
+ return;
+ }
+ if (abortAfterNanos > 0 && now - busySince >= abortAfterNanos) {
+ if (abortMessage == null) {
+ abortMessage =
+ BaseMessages.getString(
+ PKG,
+ "GoogleStorageStallWatchdog.Aborted." + direction.key,
+ uri,
+ seconds(now - busySince));
+ message = abortMessage;
+ }
+ // Again on every round until the call ends: the client may swallow
one, for example
+ // in the sleep between two attempts. One at a time, as an interrupt
can take a while.
+ if (!interrupting) {
+ interrupting = true;
+ interrupt = true;
+ }
+ } else if (now >= nextReport) {
+ nextReport = now + warnAfterNanos;
+ stalled = true;
+ message =
+ BaseMessages.getString(
+ PKG,
+ "GoogleStorageStallWatchdog.Stalled." + direction.key,
+ uri,
+ seconds(now - busySince),
+ bytes);
+ }
+ }
+ if (message != null) {
+ watchdog.report(message);
+ }
+ if (interrupt) {
+ try {
+ watchdog.interrupter.execute(this::interruptIfStillBusy);
+ } catch (RejectedExecutionException e) {
+ synchronized (this) {
+ interrupting = false;
+ }
+ }
+ }
+ }
+
+ /**
+ * Interrupt the threads, but only while they are still inside the call
the watchdog ended:
+ * holding the {@link #interruptLock} that {@link #end(long)} needs makes
sure an interrupt
+ * never reaches a thread that has moved on to something else. The
transfer's own lock is let go
+ * before the interrupt, so the checks never wait for it.
+ */
+ private void interruptIfStillBusy() {
+ synchronized (interruptLock) {
+ List<Thread> targets;
+ synchronized (this) {
+ targets = abortMessage == null ? List.of() : new
ArrayList<>(threads);
+ }
+ try {
+ for (Thread target : targets) {
+ target.interrupt();
+ }
+ } finally {
+ synchronized (this) {
+ interrupting = false;
+ }
+ }
+ }
+ }
+
+ private static long seconds(long nanos) {
+ return TimeUnit.NANOSECONDS.toSeconds(nanos);
+ }
+ }
+}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageVfsPlugin.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageVfsPlugin.java
index 934fe9026e..bbd9c8b922 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageVfsPlugin.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageVfsPlugin.java
@@ -63,7 +63,7 @@ public class GoogleStorageVfsPlugin implements IVfs {
new GoogleStorageFileProvider(variables,
googleStorageMetadataType));
}
} catch (Exception e) {
- LogChannel.GENERAL.logError("Unable to load Google Storage VFS
providers", e);
+ LogChannel.GENERAL.logError("Google Cloud Storage: Unable to load the
VFS providers", e);
}
return providers;
}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/LoggingStorageRetryStrategy.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/LoggingStorageRetryStrategy.java
new file mode 100644
index 0000000000..3e20458d7e
--- /dev/null
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/LoggingStorageRetryStrategy.java
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.gs;
+
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.retrying.TimedAttemptSettings;
+import com.google.cloud.BaseServiceException;
+import com.google.cloud.storage.StorageRetryStrategy;
+import java.io.Serializable;
+import java.net.SocketTimeoutException;
+import org.apache.hop.core.Const;
+import org.apache.hop.core.logging.LogChannel;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.vfs.gs.config.GoogleCloudConfig;
+
+/**
+ * Logs every retry the Google Cloud Storage client makes. The client retries
timeouts and temporary
+ * errors on its own and says nothing about it, so a slow or flaky connection
looks exactly like a
+ * hang. This decorates the configured strategy: the decision whether to retry
is left to it
+ * unchanged, and each time it decides to retry, a line is logged at the basic
level naming the
+ * error and, when known, the {@code gs://} object involved.
+ *
+ * <p>The client asks for a decision before it checks its attempt budget, so
the last line before
+ * giving up also says it is trying again; the error that follows says
otherwise.
+ */
+class LoggingStorageRetryStrategy implements StorageRetryStrategy {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Class<?> PKG = LoggingStorageRetryStrategy.class;
+
+ /** Where a retry is reported. Serializable because the storage options that
hold it are. */
+ @FunctionalInterface
+ interface RetryLog extends Serializable {
+ void retrying(String message);
+ }
+
+ private final Reporter reporter;
+ private final ResultRetryAlgorithm<?> idempotentHandler;
+ private final ResultRetryAlgorithm<?> nonIdempotentHandler;
+
+ LoggingStorageRetryStrategy(StorageRetryStrategy delegate, GoogleCloudConfig
config) {
+ this(
+ delegate,
+ config,
+ message -> LogChannel.GENERAL.logBasic("Google Cloud Storage: " +
message));
+ }
+
+ LoggingStorageRetryStrategy(
+ StorageRetryStrategy delegate, GoogleCloudConfig config, RetryLog
retryLog) {
+ this.reporter =
+ new Reporter(
+ retryLog,
+ Const.toInt(config.getConnectionTimeout(), 20),
+ Const.toInt(config.getReadTimeout(), 20));
+ this.idempotentHandler = reporting(delegate.getIdempotentHandler());
+ this.nonIdempotentHandler = reporting(delegate.getNonidempotentHandler());
+ }
+
+ @Override
+ public ResultRetryAlgorithm<?> getIdempotentHandler() {
+ return idempotentHandler;
+ }
+
+ @Override
+ public ResultRetryAlgorithm<?> getNonidempotentHandler() {
+ return nonIdempotentHandler;
+ }
+
+ private <T> ResultRetryAlgorithm<T> reporting(ResultRetryAlgorithm<T>
algorithm) {
+ return new Reporting<>(algorithm, reporter);
+ }
+
+ /** Answers exactly what the wrapped algorithm answers, and reports it when
that is a retry. */
+ private static final class Reporting<T> implements ResultRetryAlgorithm<T>,
Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final ResultRetryAlgorithm<T> algorithm;
+ private final Reporter reporter;
+
+ private Reporting(ResultRetryAlgorithm<T> algorithm, Reporter reporter) {
+ this.algorithm = algorithm;
+ this.reporter = reporter;
+ }
+
+ @Override
+ public TimedAttemptSettings createNextAttempt(
+ Throwable previousThrowable, T previousResponse, TimedAttemptSettings
previousSettings) {
+ return algorithm.createNextAttempt(previousThrowable, previousResponse,
previousSettings);
+ }
+
+ @Override
+ public boolean shouldRetry(Throwable previousThrowable, T
previousResponse) {
+ boolean retry = algorithm.shouldRetry(previousThrowable,
previousResponse);
+ if (retry && previousThrowable != null) {
+ reporter.report(previousThrowable);
+ }
+ return retry;
+ }
+ }
+
+ /** Turns a retried error into a line in the log. */
+ private static final class Reporter implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final RetryLog retryLog;
+ private final int connectTimeout;
+ private final int readTimeout;
+
+ private Reporter(RetryLog retryLog, int connectTimeout, int readTimeout) {
+ this.retryLog = retryLog;
+ this.connectTimeout = connectTimeout;
+ this.readTimeout = readTimeout;
+ }
+
+ void report(Throwable error) {
+ try {
+ retryLog.retrying(message(error,
GoogleStorageObjectContext.current()));
+ } catch (RuntimeException e) {
+ // This runs inside the client's retry decision: failing to log must
never change it.
+ }
+ }
+
+ private String message(Throwable error, String object) {
+ SocketTimeoutException timeout = findTimeout(error);
+ if (timeout != null) {
+ String what = Const.NVL(timeout.getMessage(), "timed out");
+ return object == null
+ ? BaseMessages.getString(
+ PKG, "LoggingStorageRetryStrategy.Timeout", what,
connectTimeout, readTimeout)
+ : BaseMessages.getString(
+ PKG,
+ "LoggingStorageRetryStrategy.Timeout.Object",
+ object,
+ what,
+ connectTimeout,
+ readTimeout);
+ }
+ String description = describe(error);
+ return object == null
+ ? BaseMessages.getString(PKG,
"LoggingStorageRetryStrategy.TemporaryError", description)
+ : BaseMessages.getString(
+ PKG, "LoggingStorageRetryStrategy.TemporaryError.Object",
object, description);
+ }
+ }
+
+ private static SocketTimeoutException findTimeout(Throwable throwable) {
+ for (Throwable t = throwable; t != null; t = t.getCause() == t ? null :
t.getCause()) {
+ if (t instanceof SocketTimeoutException) {
+ return (SocketTimeoutException) t;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * A one-line description of a temporary error. An HTTP status is reported
with the message GCS
+ * sent along; anything else - a reset connection, say - by its root cause,
since that is where
+ * the client keeps the useful part.
+ */
+ static String describe(Throwable throwable) {
+ if (throwable instanceof BaseServiceException) {
+ BaseServiceException serviceException = (BaseServiceException) throwable;
+ int code = serviceException.getCode();
+ if (code > 0) {
+ // The message often starts with the status already, as in "503
Service Unavailable".
+ String message = Const.NVL(serviceException.getMessage(), "");
+ return message.startsWith(Integer.toString(code))
+ ? "HTTP " + message
+ : "HTTP " + code + " " + message;
+ }
+ }
+ Throwable root = throwable;
+ while (root.getCause() != null && root.getCause() != root) {
+ root = root.getCause();
+ }
+ String name = root.getClass().getSimpleName();
+ return root.getMessage() == null ? name : name + ": " + root.getMessage();
+ }
+}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ReadChannelInputStream.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ReadChannelInputStream.java
index 42021371e2..59f63d1dd8 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ReadChannelInputStream.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ReadChannelInputStream.java
@@ -28,8 +28,9 @@ public class ReadChannelInputStream extends InputStream {
/**
* Volatile and intentionally NOT guarded by the read monitor: {@link
#close()} must be able to
* tear the channel down from another thread (e.g. a pipeline stop) even
while {@code read()} is
- * parked inside {@code channel.read(...)} on a stalled network socket.
Closing the channel from
- * the other thread aborts the in-flight read, which is the only way to
break a stalled transfer.
+ * parked inside {@code channel.read(...)} on a stalled network socket. Note
that the GCS client's
+ * own {@code close()} waits for a read in progress, so this does not break
a stalled read; see
+ * {@code GoogleStorageStalledDownloadTest}.
*/
private volatile ReadChannel channel;
@@ -40,8 +41,26 @@ public class ReadChannelInputStream extends InputStream {
private final ByteBuffer bytes = ByteBuffer.allocate(64 * 1024);
+ /** The object being read, named in the retry log and watched for stalls. */
+ private final GoogleStorageStallWatchdog.Transfer transfer;
+
+ /**
+ * Set once the watchdog ended a stuck read. Every later read fails with it
too: the download is
+ * incomplete, and returning end-of-stream would silently cut the file short.
+ */
+ private volatile IOException failure;
+
public ReadChannelInputStream(ReadChannel channel) {
+ this(channel, GoogleStorageStallWatchdog.Transfer.untracked(null));
+ }
+
+ /**
+ * @param channel the channel to read from
+ * @param transfer the object being read
+ */
+ ReadChannelInputStream(ReadChannel channel,
GoogleStorageStallWatchdog.Transfer transfer) {
this.channel = channel;
+ this.transfer = transfer;
}
@Override
@@ -58,6 +77,9 @@ public class ReadChannelInputStream extends InputStream {
public int read(byte[] buf, int off, int len) throws IOException {
// Snapshot the channel; close() may null it concurrently and that is
allowed - it is how a
// stalled read gets unblocked.
+ if (failure != null) {
+ throw new IOException(failure.getMessage(), failure);
+ }
ReadChannel ch = channel;
if (ch == null) {
return -1;
@@ -67,16 +89,39 @@ public class ReadChannelInputStream extends InputStream {
if (len < bytes.remaining()) {
bytes.limit(len);
}
- int res;
+ int res = -1;
+ boolean returned = false;
+ boolean aborted = false;
+ transfer.begin();
try {
- res = ch.read(bytes);
+ try (GoogleStorageObjectContext.Scope ignored =
+ GoogleStorageObjectContext.enter(transfer.uri())) {
+ res = ch.read(bytes);
+ returned = true;
+ } finally {
+ // Also on an Error: a transfer left busy could have its thread
interrupted later on.
+ // Ended by the watchdog just as the data arrived: the read
succeeded, so carry on.
+ aborted = returned ? transfer.end(res) : transfer.failed();
+ }
} catch (IOException e) {
+ if (aborted) {
+ // The watchdog interrupted a read that made no progress for too
long.
+ failure = transfer.abortedError(e);
+ throw failure;
+ }
// If close() ran underneath us the read fails - treat that as a
normal end-of-stream
// rather than propagating a spurious error from the deliberate
teardown.
if (channel == null) {
return -1;
}
throw e;
+ } catch (RuntimeException e) {
+ // However the client reports the interrupt, the reason it came is the
watchdog's.
+ if (aborted) {
+ failure = transfer.abortedError(e);
+ throw failure;
+ }
+ throw e;
}
if (res < 0) {
close();
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/WriteChannelOutputStream.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/WriteChannelOutputStream.java
index 48ef81b2b5..1d029de3ea 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/WriteChannelOutputStream.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/WriteChannelOutputStream.java
@@ -48,8 +48,20 @@ public class WriteChannelOutputStream extends OutputStream {
private final ByteBuffer bytes = ByteBuffer.allocate(64 * 1024);
+ /** The object being written, named in the retry log and watched for stalls.
*/
+ private final GoogleStorageStallWatchdog.Transfer transfer;
+
public WriteChannelOutputStream(WriteChannel channel) {
+ this(channel, GoogleStorageStallWatchdog.Transfer.untracked(null));
+ }
+
+ /**
+ * @param channel the channel to write to
+ * @param transfer the object being written
+ */
+ WriteChannelOutputStream(WriteChannel channel,
GoogleStorageStallWatchdog.Transfer transfer) {
this.channel = channel;
+ this.transfer = transfer;
}
@Override
@@ -63,32 +75,47 @@ public class WriteChannelOutputStream extends OutputStream {
throw new ClosedChannelException();
}
synchronized (writeLock) {
- int count = 0;
- while (count < len) {
- int c = Math.min(len - count, bytes.remaining());
- bytes.put(buf, off + count, c);
- bytes.flip();
- int noProgress = 0;
- while (bytes.hasRemaining()) {
- WriteChannel ch = channel;
- if (ch == null) {
- // close() ran on another thread (e.g. pipeline stop) - stop
writing.
- throw new ClosedChannelException();
- }
- int written = ch.write(bytes);
- if (written <= 0) {
- if (++noProgress > MAX_NO_PROGRESS_WRITES) {
- throw new IOException(
- "Google Storage write channel made no progress after "
- + MAX_NO_PROGRESS_WRITES
- + " attempts; aborting to avoid an unbounded loop");
+ try (GoogleStorageObjectContext.Scope ignored =
+ GoogleStorageObjectContext.enter(transfer.uri())) {
+ int count = 0;
+ while (count < len) {
+ int c = Math.min(len - count, bytes.remaining());
+ bytes.put(buf, off + count, c);
+ bytes.flip();
+ int noProgress = 0;
+ while (bytes.hasRemaining()) {
+ WriteChannel ch = channel;
+ if (ch == null) {
+ // close() ran on another thread (e.g. pipeline stop) - stop
writing.
+ throw new ClosedChannelException();
+ }
+ int written = 0;
+ boolean returned = false;
+ transfer.begin();
+ try {
+ written = ch.write(bytes);
+ returned = true;
+ } finally {
+ if (returned) {
+ transfer.end(written);
+ } else {
+ transfer.failed();
+ }
+ }
+ if (written <= 0) {
+ if (++noProgress > MAX_NO_PROGRESS_WRITES) {
+ throw new IOException(
+ "Google Storage write channel made no progress after "
+ + MAX_NO_PROGRESS_WRITES
+ + " attempts; aborting to avoid an unbounded loop");
+ }
+ } else {
+ noProgress = 0;
}
- } else {
- noProgress = 0;
}
+ bytes.compact();
+ count += c;
}
- bytes.compact();
- count += c;
}
}
}
@@ -100,7 +127,21 @@ public class WriteChannelOutputStream extends OutputStream
{
WriteChannel ch = channel;
if (ch != null) {
channel = null;
- ch.close();
+ // The last part of the upload happens here, possibly while a write() is
still inside the
+ // client: the transfer counts both calls and stays watched until the
last one is done.
+ boolean returned = false;
+ transfer.begin();
+ try (GoogleStorageObjectContext.Scope ignored =
+ GoogleStorageObjectContext.enter(transfer.uri())) {
+ ch.close();
+ returned = true;
+ } finally {
+ if (returned) {
+ transfer.end(0);
+ } else {
+ transfer.failed();
+ }
+ }
}
}
}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigPlugin.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigPlugin.java
index 5f11a9ceb6..576ec74208 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigPlugin.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigPlugin.java
@@ -248,7 +248,7 @@ public class GoogleCloudConfigPlugin implements
IConfigOptions, IGuiPluginCompos
if (serviceAccountKeyFile != null) {
config.setServiceAccountKeyFile(serviceAccountKeyFile);
log.logBasic(
- "The Google Cloud service account JSON jey file is set to '"
+ "Google Cloud Storage: the service account JSON key file is set to
'"
+ serviceAccountKeyFile
+ "'");
@@ -258,81 +258,81 @@ public class GoogleCloudConfigPlugin implements
IConfigOptions, IGuiPluginCompos
if (scanFoldersForLastModifDate != null &&
scanFoldersForLastModifDate.equals(Boolean.TRUE)) {
config.setScanFoldersForLastModifDate(scanFoldersForLastModifDate);
log.logBasic(
- "Google Cloud Storage service will scan folders for the last file
modification time.");
+ "Google Cloud Storage: folders will be scanned for the last file
modification time.");
changed = true;
}
if (maxAttempts != null) {
config.setMaxAttempts(maxAttempts);
- log.logBasic("Google Cloud service max attempts set to " +
maxAttempts);
+ log.logBasic("Google Cloud Storage: max attempts set to " +
maxAttempts);
changed = true;
}
if (initialRetryDelay != null) {
config.setInitialRetryDelay(initialRetryDelay);
- log.logBasic("Google Cloud service initialRetryDelay set to " +
initialRetryDelay);
+ log.logBasic("Google Cloud Storage: initialRetryDelay set to " +
initialRetryDelay);
changed = true;
}
if (retryDelayMultiplier != null) {
config.setRetryDelayMultiplier(retryDelayMultiplier);
- log.logBasic("Google Cloud service retryDelayMultiplier set to " +
retryDelayMultiplier);
+ log.logBasic("Google Cloud Storage: retryDelayMultiplier set to " +
retryDelayMultiplier);
changed = true;
}
if (maxRetryDelay != null) {
config.setMaxRetryDelay(maxRetryDelay);
- log.logBasic("Google Cloud service maxRetryDelay set to " +
maxRetryDelay);
+ log.logBasic("Google Cloud Storage: maxRetryDelay set to " +
maxRetryDelay);
changed = true;
}
if (totalTimeout != null) {
config.setTotalTimeout(totalTimeout);
- log.logBasic("Google Cloud service totalTimeout set to " +
totalTimeout);
+ log.logBasic("Google Cloud Storage: totalTimeout set to " +
totalTimeout);
changed = true;
}
if (initialRpcTimeout != null) {
config.setInitialRpcTimeout(initialRpcTimeout);
- log.logBasic("Google Cloud service initialRpcTimeout set to " +
initialRpcTimeout);
+ log.logBasic("Google Cloud Storage: initialRpcTimeout set to " +
initialRpcTimeout);
changed = true;
}
if (rpcTimeoutMultiplier != null) {
config.setRpcTimeoutMultiplier(rpcTimeoutMultiplier);
- log.logBasic("Google Cloud service rpcTimeoutMultiplier set to " +
rpcTimeoutMultiplier);
+ log.logBasic("Google Cloud Storage: rpcTimeoutMultiplier set to " +
rpcTimeoutMultiplier);
changed = true;
}
if (maxRpcTimeout != null) {
config.setMaxRpcTimeout(maxRpcTimeout);
- log.logBasic("Google Cloud service maxRpcTimeout set to " +
maxRpcTimeout);
+ log.logBasic("Google Cloud Storage: maxRpcTimeout set to " +
maxRpcTimeout);
changed = true;
}
if (connectionTimeout != null) {
config.setConnectionTimeout(connectionTimeout);
- log.logBasic("Google Cloud service connectionTimeout set to " +
connectionTimeout);
+ log.logBasic("Google Cloud Storage: connectionTimeout set to " +
connectionTimeout);
changed = true;
}
if (readTimeout != null) {
config.setReadTimeout(readTimeout);
- log.logBasic("Google Cloud service readTimeout set to " + readTimeout);
+ log.logBasic("Google Cloud Storage: readTimeout set to " +
readTimeout);
changed = true;
}
if (retryNonIdempotentOperations != null) {
config.setRetryNonIdempotentOperations(retryNonIdempotentOperations);
log.logBasic(
- "Google Cloud service retry of non-idempotent operations set to "
+ "Google Cloud Storage: retry of non-idempotent operations set to "
+ retryNonIdempotentOperations);
changed = true;
}
if (cacheTtlSeconds != null) {
config.setCacheTtlSeconds(cacheTtlSeconds);
- log.logBasic("Google Cloud list cache TTL (seconds) set to " +
cacheTtlSeconds);
+ log.logBasic("Google Cloud Storage: list cache TTL (seconds) set to "
+ cacheTtlSeconds);
changed = true;
}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigSingleton.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigSingleton.java
index 6433c07a85..e3a3f2214f 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigSingleton.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/config/GoogleCloudConfigSingleton.java
@@ -50,7 +50,7 @@ public class GoogleCloudConfigSingleton {
mapper.readValue(new Gson().toJson(configObject),
GoogleCloudConfig.class);
} catch (Exception e) {
LogChannel.GENERAL.logError(
- "Error reading Google Drive configuration, check property '"
+ "Google Cloud Storage: Error reading the Google Cloud
configuration, check property '"
+ GoogleCloudConfig.HOP_CONFIG_GOOGLE_CLOUD_CONFIG_KEY
+ "' in the Hop config json file",
e);
diff --git
a/plugins/tech/google/src/main/resources/org/apache/hop/vfs/gs/messages/messages_en_US.properties
b/plugins/tech/google/src/main/resources/org/apache/hop/vfs/gs/messages/messages_en_US.properties
new file mode 100644
index 0000000000..7b6d5a6297
--- /dev/null
+++
b/plugins/tech/google/src/main/resources/org/apache/hop/vfs/gs/messages/messages_en_US.properties
@@ -0,0 +1,28 @@
+#
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+
+LoggingStorageRetryStrategy.Timeout=No response in time ({0}), trying again.
The connection timeout is {1} seconds, the read timeout {2} seconds.
+LoggingStorageRetryStrategy.Timeout.Object=No response in time while accessing
{0} ({1}), trying again. The connection timeout is {2} seconds, the read
timeout {3} seconds.
+LoggingStorageRetryStrategy.TemporaryError=Temporary error, trying again: {0}
+LoggingStorageRetryStrategy.TemporaryError.Object=Temporary error while
accessing {0}, trying again: {1}
+GoogleStorageStallWatchdog.Stalled.Reading=No progress reading {0} for {1}
seconds ({2} bytes read so far), still waiting.
+GoogleStorageStallWatchdog.Stalled.Writing=Writing {0} has not finished an
upload chunk for {1} seconds ({2} bytes written so far): the connection is slow
or stalled.
+GoogleStorageStallWatchdog.Resumed.Reading=Reading {0} resumed after {1}
seconds without progress.
+GoogleStorageStallWatchdog.Resumed.Writing=Writing {0} finished the upload
chunk after {1} seconds.
+GoogleStorageStallWatchdog.Aborted.Reading=No progress reading {0} for {1}
seconds, giving up: that is as long as the Google Cloud retry settings allow
for one read.
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java
index 79b734940e..3e104cca34 100644
---
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java
@@ -38,6 +38,10 @@ import com.google.cloud.storage.StorageException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
@@ -110,6 +114,49 @@ class ComposeAppendOutputStreamTest {
inOrder.verify(storage).delete(BlobId.of(BUCKET, TEMP));
}
+ /** The upload of the appended bytes is watched, and reported under the name
of the target. */
+ @Test
+ void aStalledAppendUploadIsReportedForTheTarget() throws Exception {
+ long[] now = {0};
+ List<String> reported = new ArrayList<>();
+ GoogleStorageStallWatchdog watchdog =
+ new GoogleStorageStallWatchdog(reported::add, () -> now[0], null,
Runnable::run);
+ GoogleStorageStallWatchdog.Limits limits =
+ new GoogleStorageStallWatchdog.Limits(
+ Duration.ofSeconds(60), Duration.ofSeconds(150),
Duration.ofSeconds(60));
+ WriteChannel tempChannel = mock(WriteChannel.class);
+ when(tempChannel.write(any(ByteBuffer.class)))
+ .thenAnswer(
+ inv -> {
+ now[0] += TimeUnit.SECONDS.toNanos(60);
+ watchdog.check();
+ ByteBuffer b = inv.getArgument(0);
+ int remaining = b.remaining();
+ b.position(b.limit());
+ return remaining;
+ });
+ String target = "gs://" + BUCKET + "/" + TARGET;
+
+ try (ComposeAppendOutputStream out =
+ new ComposeAppendOutputStream(
+ mock(Storage.class),
+ BUCKET,
+ TARGET,
+ TEMP,
+ GENERATION,
+ tempChannel,
+ watchdog.writing(target, limits))) {
+ out.write("appended".getBytes(StandardCharsets.UTF_8));
+ }
+
+ assertEquals(
+ "Writing "
+ + target
+ + " has not finished an upload chunk for 60 seconds (0 bytes
written so far): the"
+ + " connection is slow or stalled.",
+ reported.get(0));
+ }
+
@Test
void composeFailurePropagatesButTempIsStillDeleted() throws Exception {
Storage storage = mock(Storage.class);
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageCredentialsTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageCredentialsTest.java
new file mode 100644
index 0000000000..47d2fc0dfe
--- /dev/null
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageCredentialsTest.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.gs;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.auth.oauth2.ServiceAccountCredentials;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.commons.vfs2.FileSystemOptions;
+import org.apache.commons.vfs2.FileType;
+import org.apache.hop.core.logging.HopLogStore;
+import org.apache.hop.core.logging.HopLoggingEvent;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.vfs.gs.config.GoogleCloudConfig;
+import org.apache.hop.vfs.gs.config.GoogleCloudConfigSingleton;
+import org.apache.hop.vfs.gs.metadatatype.GoogleStorageCredentialsType;
+import org.apache.hop.vfs.gs.metadatatype.GoogleStorageMetadataType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentMatchers;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+/**
+ * Credentials that cannot be loaded used to surface as a bare
NullPointerException from the Google
+ * client on the first {@code gs://} access, and for the default scheme
nothing was logged at all.
+ * Now the first use fails with an error that says why, logged once.
+ */
+class GoogleStorageCredentialsTest {
+
+ private static final String MISSING_KEY_FILE =
"/nonexistent/hop-gcs-credentials-test-key.json";
+
+ private String originalKeyFile;
+
+ @BeforeAll
+ static void initLogging() {
+ HopLogStore.init();
+ }
+
+ @BeforeEach
+ void rememberKeyFile() {
+ originalKeyFile =
GoogleCloudConfigSingleton.getConfig().getServiceAccountKeyFile();
+ }
+
+ @AfterEach
+ void restoreKeyFile() {
+
GoogleCloudConfigSingleton.getConfig().setServiceAccountKeyFile(originalKeyFile);
+ }
+
+ @Test
+ void aKeyFileThatCannotBeReadFailsClearlyOnFirstUse() throws Exception {
+ GoogleCloudConfig config = GoogleCloudConfigSingleton.getConfig();
+ config.setServiceAccountKeyFile(MISSING_KEY_FILE);
+
+ GoogleStorageFileSystem fileSystem = fileSystem(new
GoogleStorageFileProvider(), "gs");
+
+ IOException error = assertThrows(IOException.class,
fileSystem::setupStorage);
+ String message = error.getMessage();
+ assertTrue(
+ message.startsWith("Google Cloud Storage: no credentials to access
gs://: "), message);
+ assertTrue(
+ message.contains("the service account key file '" + MISSING_KEY_FILE +
"'"), message);
+ assertTrue(message.contains("FileNotFoundException"), message);
+
+ // Every use fails the same way, but the log says it only once.
+ assertThrows(IOException.class, fileSystem::setupStorage);
+ assertThrows(IOException.class, fileSystem::getStorageControlClient);
+ assertEquals(1, loggedLinesContaining(MISSING_KEY_FILE));
+ }
+
+ /**
+ * A missing dependency while reading the key file used to be reported as
missing Application
+ * Default Credentials, telling the user to set the key file they had
already set.
+ */
+ @Test
+ void aLinkageErrorReadingTheKeyFileBlamesTheKeyFile(@TempDir Path folder)
throws Exception {
+ Path keyFile = Files.writeString(folder.resolve("key.json"), "{}");
+
GoogleCloudConfigSingleton.getConfig().setServiceAccountKeyFile(keyFile.toString());
+
+ GoogleStorageFileSystem fileSystem;
+ try (MockedStatic<ServiceAccountCredentials> credentials =
+ Mockito.mockStatic(ServiceAccountCredentials.class)) {
+ credentials
+ .when(() ->
ServiceAccountCredentials.fromStream(ArgumentMatchers.any(InputStream.class)))
+ .thenThrow(new NoClassDefFoundError("io/grpc/Context"));
+ fileSystem = fileSystem(new GoogleStorageFileProvider(), "gs");
+ }
+
+ String message = assertThrows(IOException.class,
fileSystem::setupStorage).getMessage();
+ assertTrue(message.contains("the service account key file '" + keyFile +
"'"), message);
+ assertTrue(message.contains("NoClassDefFoundError: io/grpc/Context"),
message);
+ assertFalse(message.contains("Application Default Credentials"), message);
+ }
+
+ @Test
+ void aNamedConnectionWithBrokenCredentialsFailsClearly() {
+ GoogleStorageMetadataType connection = new GoogleStorageMetadataType();
+ connection.setName("broken-gcs");
+
connection.setStorageCredentialsType(GoogleStorageCredentialsType.KEY_STRING);
+ connection.setStorageAccountKey("this is not a service account key");
+
+ GoogleStorageFileSystem fileSystem =
+ fileSystem(new GoogleStorageFileProvider(new Variables(), connection),
"broken-gcs");
+
+ IOException error = assertThrows(IOException.class,
fileSystem::setupStorage);
+ String message = error.getMessage();
+ assertTrue(
+ message.startsWith("Google Cloud Storage: no credentials to access
broken-gcs://: "),
+ message);
+ assertTrue(
+ message.contains(
+ "the credentials of Google Storage connection 'broken-gcs' could
not be loaded"),
+ message);
+ }
+
+ private static GoogleStorageFileSystem fileSystem(
+ GoogleStorageFileProvider provider, String scheme) {
+ try {
+ return (GoogleStorageFileSystem)
+ provider.doCreateFileSystem(
+ new GoogleStorageFileName(scheme, "/", FileType.FOLDER), new
FileSystemOptions());
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static long loggedLinesContaining(String text) {
+ return HopLogStore.getAppender()
+ .getLogBufferFromTo((java.util.List<String>) null, true, 0,
Integer.MAX_VALUE)
+ .stream()
+ .map(HopLoggingEvent::getMessage)
+ .map(String::valueOf)
+ .filter(line -> line.contains(text))
+ .count();
+ }
+}
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageRetryLoggingTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageRetryLoggingTest.java
new file mode 100644
index 0000000000..0d4199dcca
--- /dev/null
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageRetryLoggingTest.java
@@ -0,0 +1,267 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.gs;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.cloud.NoCredentials;
+import com.google.cloud.http.HttpTransportOptions;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.hop.vfs.gs.config.GoogleCloudConfig;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.threeten.bp.Duration;
+
+/**
+ * The storage client retries timeouts and temporary errors without telling
anyone, which makes a
+ * flaky connection indistinguishable from a hang. Each retry is now logged;
these tests check what
+ * a user gets to see, against a local endpoint and the real client.
+ */
+class GoogleStorageRetryLoggingTest {
+
+ private static final String URI = "gs://bucket/file.txt";
+
+ private static final byte[] CONTENT =
"id;name\n1;one\n".getBytes(StandardCharsets.UTF_8);
+
+ private static final int READ_TIMEOUT_MS = 250;
+
+ private static final String UNAVAILABLE =
+ "{\"error\":{\"code\":503,\"message\":\"Backend
Error\",\"errors\":[{\"domain\":\"global\","
+ + "\"reason\":\"backendError\",\"message\":\"Backend Error\"}]}}";
+
+ private static final String OBJECT =
+
"{\"kind\":\"storage#object\",\"bucket\":\"bucket\",\"name\":\"file.txt\","
+ + "\"generation\":\"1\",\"metageneration\":\"1\",\"size\":\"14\"}";
+
+ private static final String NOT_FOUND =
+ "{\"error\":{\"code\":404,\"message\":\"No such object:
bucket/file.txt\",\"errors\":"
+ + "[{\"domain\":\"global\",\"reason\":\"notFound\",\"message\":\"No
such object\"}]}}";
+
+ private enum Answer {
+ CONTENT,
+ UNAVAILABLE,
+ NOT_FOUND,
+ STALL_AFTER_HEADERS
+ }
+
+ private HttpServer server;
+ private ExecutorService handlers;
+
+ /** The answers to the next requests, in order; once used up the object is
served normally. */
+ private final Queue<Answer> script = new ConcurrentLinkedQueue<>();
+
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final List<String> logged = new CopyOnWriteArrayList<>();
+
+ @BeforeEach
+ void startEndpoint() throws IOException {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ handlers =
+ Executors.newCachedThreadPool(
+ r -> {
+ Thread t = new Thread(r, "fake-gcs-handler");
+ t.setDaemon(true);
+ return t;
+ });
+ server.setExecutor(handlers);
+ server.createContext("/", this::handle);
+ server.start();
+ }
+
+ @AfterEach
+ void stopEndpoint() {
+ release.countDown();
+ server.stop(0);
+ handlers.shutdownNow();
+ }
+
+ private void handle(HttpExchange exchange) throws IOException {
+ Answer answer = script.poll();
+ switch (answer == null ? Answer.CONTENT : answer) {
+ case CONTENT:
+ if
(!String.valueOf(exchange.getRequestURI().getQuery()).contains("alt=media")) {
+ // A metadata call rather than a download.
+ json(exchange, 200, OBJECT);
+ break;
+ }
+ objectHeaders(exchange);
+ exchange.sendResponseHeaders(200, CONTENT.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(CONTENT);
+ }
+ break;
+ case UNAVAILABLE:
+ json(exchange, 503, UNAVAILABLE);
+ break;
+ case NOT_FOUND:
+ json(exchange, 404, NOT_FOUND);
+ break;
+ case STALL_AFTER_HEADERS:
+ objectHeaders(exchange);
+ exchange.sendResponseHeaders(200, CONTENT.length);
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ exchange.close();
+ break;
+ }
+ }
+
+ private static void objectHeaders(HttpExchange exchange) {
+ exchange.getResponseHeaders().add("Content-Type", "text/plain");
+ exchange.getResponseHeaders().add("x-goog-generation", "1");
+ exchange.getResponseHeaders().add("x-goog-metageneration", "1");
+ }
+
+ private static void json(HttpExchange exchange, int code, String json)
throws IOException {
+ byte[] body = json.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/json;
charset=UTF-8");
+ exchange.sendResponseHeaders(code, body.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(body);
+ }
+ }
+
+ @Test
+ void eachRetryOfATemporaryErrorIsLoggedWithTheObject() throws Exception {
+ script.add(Answer.UNAVAILABLE);
+ script.add(Answer.UNAVAILABLE);
+
+ assertArrayEquals(CONTENT, read(storage(this::log)));
+
+ assertEquals(2, logged.size(), "one line per retry: " + logged);
+ for (String line : logged) {
+ assertTrue(line.contains("Temporary error while accessing " + URI),
line);
+ assertTrue(line.contains("HTTP 503"), line);
+ assertTrue(line.contains("trying again"), line);
+ }
+ }
+
+ @Test
+ void aTimeoutIsLoggedAsATimeout() throws Exception {
+ script.add(Answer.STALL_AFTER_HEADERS);
+
+ assertArrayEquals(CONTENT, read(storage(this::log)));
+
+ assertEquals(1, logged.size(), "one line for the one timeout: " + logged);
+ String line = logged.get(0);
+ assertTrue(line.contains("No response in time while accessing " + URI),
line);
+ assertTrue(line.contains("Read timed out"), line);
+ assertTrue(line.contains("the read timeout 20 seconds"), line);
+ }
+
+ @Test
+ void anErrorThatIsNotRetriedIsNotLogged() {
+ script.add(Answer.NOT_FOUND);
+
+ assertThrows(IOException.class, () -> read(storage(this::log)));
+ assertEquals(List.of(), logged);
+ }
+
+ @Test
+ void aCallOutsideAnyObjectIsLoggedWithoutOne() {
+ script.add(Answer.UNAVAILABLE);
+
+ storage(this::log).get(BlobId.of("bucket", "file.txt"));
+
+ assertEquals(1, logged.size(), "one line per retry: " + logged);
+ String line = logged.get(0);
+ assertTrue(line.startsWith("Temporary error, trying again: HTTP 503"),
line);
+ }
+
+ /** The log is called from inside the client's retry decision; it must not
be able to break it. */
+ @Test
+ void aFailingLogDoesNotChangeTheRetries() throws Exception {
+ script.add(Answer.UNAVAILABLE);
+ script.add(Answer.UNAVAILABLE);
+
+ Storage storage =
+ storage(
+ message -> {
+ throw new IllegalStateException("the logging system is not
initialized");
+ });
+
+ assertArrayEquals(CONTENT, read(storage));
+ }
+
+ private void log(String message) {
+ logged.add(message);
+ }
+
+ private static byte[] read(Storage storage) throws IOException {
+ try (InputStream in =
+ new ReadChannelInputStream(
+ storage.reader(BlobId.of("bucket", "file.txt")),
+ GoogleStorageStallWatchdog.Transfer.untracked(URI))) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ in.transferTo(out);
+ return out.toByteArray();
+ }
+ }
+
+ /**
+ * Builds the client the way {@link GoogleStorageFileSystem#setupStorage()}
does, pointed at the
+ * local endpoint, with retry delays and timeouts shortened to keep the test
fast. The timeouts
+ * named in the log come from the configuration, which keeps its defaults.
+ */
+ private Storage storage(LoggingStorageRetryStrategy.RetryLog retryLog) {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ RetrySettings prompt =
+ GoogleStorageFileSystem.buildRetrySettings(config).toBuilder()
+ .setInitialRetryDelay(Duration.ofMillis(1))
+ .setMaxRetryDelay(Duration.ofMillis(2))
+ .build();
+
+ return GoogleStorageFileSystem.buildStorageOptions(config, retryLog)
+ .setRetrySettings(prompt)
+ .setTransportOptions(
+ HttpTransportOptions.newBuilder()
+ .setConnectTimeout(READ_TIMEOUT_MS)
+ .setReadTimeout(READ_TIMEOUT_MS)
+ .build())
+ .setHost("http://127.0.0.1:"; + server.getAddress().getPort())
+ .setProjectId("hop-test")
+ .setCredentials(NoCredentials.getInstance())
+ .build()
+ .getService();
+ }
+}
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdogTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdogTest.java
new file mode 100644
index 0000000000..845b2a531f
--- /dev/null
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdogTest.java
@@ -0,0 +1,516 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.gs;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.hop.vfs.gs.config.GoogleCloudConfig;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Which transfers the watchdog reports or ends, and when - on a clock the
test moves by hand. The
+ * transfers run on the test thread, so an interrupt from the watchdog lands
on the test thread too.
+ */
+class GoogleStorageStallWatchdogTest {
+
+ private static final String URI = "gs://bucket/file.txt";
+
+ /** Report after a minute, end a read after two and a half, report a write
after a minute. */
+ private static final GoogleStorageStallWatchdog.Limits LIMITS =
+ new GoogleStorageStallWatchdog.Limits(
+ Duration.ofSeconds(60), Duration.ofSeconds(150),
Duration.ofSeconds(60));
+
+ private long now = TimeUnit.HOURS.toNanos(1);
+ private final List<String> logged = new ArrayList<>();
+ private final GoogleStorageStallWatchdog watchdog =
+ new GoogleStorageStallWatchdog(logged::add, () -> now, null,
Runnable::run);
+
+ @AfterEach
+ void clearInterrupt() {
+ Thread.interrupted();
+ }
+
+ @Test
+ void aTransferThatKeepsMovingIsNotReported() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+
+ for (int i = 0; i < 10; i++) {
+ transfer.begin();
+ advance(59);
+ watchdog.check();
+ transfer.end(100);
+ }
+
+ assertEquals(List.of(), logged);
+ }
+
+ @Test
+ void aBlockedReadIsReportedOncePerWarningPeriod() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+ transfer.begin();
+ transfer.end(512);
+
+ transfer.begin();
+ advance(59);
+ watchdog.check();
+ assertEquals(List.of(), logged, "not yet");
+
+ advance(1);
+ watchdog.check();
+ advance(1);
+ watchdog.check();
+ assertEquals(
+ List.of(
+ "No progress reading "
+ + URI
+ + " for 60 seconds (512 bytes read so far), still waiting."),
+ logged);
+
+ advance(59);
+ watchdog.check();
+ assertEquals(2, logged.size());
+ assertEquals(
+ "No progress reading " + URI + " for 120 seconds (512 bytes read so
far), still waiting.",
+ logged.get(1));
+ assertFalse(Thread.currentThread().isInterrupted(), "reported, not ended");
+ }
+
+ @Test
+ void aReadStuckForTheAbortPeriodIsEnded() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+ transfer.begin();
+ advance(149);
+ watchdog.check();
+ assertFalse(Thread.currentThread().isInterrupted(), "not yet");
+
+ advance(1);
+ watchdog.check();
+
+ assertTrue(Thread.currentThread().isInterrupted(), "the reading thread is
interrupted");
+ String aborted =
+ "No progress reading "
+ + URI
+ + " for 150 seconds, giving up: that is as long as the Google
Cloud retry settings"
+ + " allow for one read.";
+ assertEquals(aborted, logged.get(logged.size() - 1));
+
+ assertTrue(transfer.end(0), "the call reports that it was ended");
+ assertFalse(
+ Thread.currentThread().isInterrupted(), "the interrupt must not
outlive the call it ended");
+ IOException error = transfer.abortedError(new IOException("closed by
interrupt"));
+ assertEquals(aborted, error.getMessage());
+ }
+
+ /** The client can swallow an interrupt, in the sleep between two attempts
for example. */
+ @Test
+ void anEndedReadIsInterruptedAgainUntilItStops() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+ transfer.begin();
+ advance(150);
+ watchdog.check();
+ assertTrue(Thread.interrupted(), "interrupted, and the client swallows
it");
+
+ advance(1);
+ watchdog.check();
+ assertTrue(Thread.currentThread().isInterrupted(), "interrupted again");
+ assertEquals(
+ 1, logged.stream().filter(line -> line.contains("giving up")).count(),
"said only once");
+
+ assertTrue(transfer.end(0));
+ }
+
+ @Test
+ void aWriteIsReportedAsSlowOrStalledButNeverEnded() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.writing(URI,
LIMITS);
+ transfer.begin();
+ advance(3600);
+ watchdog.check();
+
+ assertFalse(Thread.currentThread().isInterrupted(), "a write is never
interrupted");
+ assertEquals(
+ List.of(
+ "Writing "
+ + URI
+ + " has not finished an upload chunk for 3600 seconds (0 bytes
written so far):"
+ + " the connection is slow or stalled."),
+ logged);
+ assertFalse(transfer.end(65536));
+ }
+
+ @Test
+ void aSlowWriteThatFinishesSaysSo() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.writing(URI,
LIMITS);
+ transfer.begin();
+ advance(60);
+ watchdog.check();
+ advance(15);
+ transfer.end(65536);
+
+ assertEquals(
+ List.of(
+ "Writing "
+ + URI
+ + " has not finished an upload chunk for 60 seconds (0 bytes
written so far):"
+ + " the connection is slow or stalled.",
+ "Writing " + URI + " finished the upload chunk after 75 seconds."),
+ logged);
+
+ // A later call that is quick again says nothing.
+ transfer.begin();
+ advance(1);
+ watchdog.check();
+ transfer.end(65536);
+ assertEquals(2, logged.size());
+ }
+
+ @Test
+ void aStalledReadThatGetsGoingAgainSaysSo() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+ transfer.begin();
+ advance(60);
+ watchdog.check();
+ advance(15);
+
+ assertFalse(transfer.end(65536));
+ assertEquals("Reading " + URI + " resumed after 75 seconds without
progress.", logged.get(1));
+ }
+
+ /**
+ * Backpressure: a transform that cannot hand its rows on stops reading for
as long as the next
+ * transform needs. That time is spent outside the client, so it is not a
stall.
+ */
+ @Test
+ void timeBetweenReadsIsNotAStall() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+ transfer.begin();
+ transfer.end(65536);
+
+ advance(3600);
+ watchdog.check();
+
+ transfer.begin();
+ advance(1);
+ watchdog.check();
+ transfer.end(65536);
+
+ assertEquals(List.of(), logged);
+ assertFalse(Thread.currentThread().isInterrupted());
+ }
+
+ /** The watchdog can hold a transfer that has just finished; it must not act
on that one. */
+ @Test
+ void aTransferThatFinishedIsLeftAlone() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+ transfer.begin();
+ advance(30);
+ transfer.end(10);
+ advance(3600);
+ watchdog.check();
+
+ assertEquals(List.of(), logged);
+ assertFalse(Thread.currentThread().isInterrupted());
+ }
+
+ @Test
+ void anUntrackedTransferIsNeverReported() {
+ GoogleStorageStallWatchdog.Transfer transfer =
+ GoogleStorageStallWatchdog.Transfer.untracked(URI);
+ transfer.begin();
+ advance(3600);
+ watchdog.check();
+
+ assertFalse(transfer.end(0));
+ assertEquals(List.of(), logged);
+ assertEquals(URI, transfer.uri());
+ }
+
+ @Test
+ void aFailingLogDoesNotStopTheWatchdog() {
+ GoogleStorageStallWatchdog failing =
+ new GoogleStorageStallWatchdog(
+ message -> {
+ throw new IllegalStateException("the logging system is not
initialized");
+ },
+ () -> now,
+ null,
+ Runnable::run);
+ GoogleStorageStallWatchdog.Transfer transfer = failing.reading(URI,
LIMITS);
+ transfer.begin();
+ advance(60);
+
+ assertDoesNotThrow(failing::check);
+ advance(90);
+ assertDoesNotThrow(failing::check);
+ assertTrue(Thread.currentThread().isInterrupted(), "a failing log does not
stop the abort");
+ assertTrue(transfer.end(0));
+ }
+
+ /**
+ * Interrupting a read blocked in a channel closes that channel on the
interrupting thread, and
+ * closing a socket can take as long as a read timeout. Meanwhile the other
transfers must still
+ * be checked.
+ */
+ @Test
+ void aSlowInterruptDoesNotHoldUpTheOtherTransfers() throws Exception {
+ CountDownLatch reading = new CountDownLatch(1);
+ CountDownLatch closing = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ // Like a socket: neither the read nor the close gives way to an interrupt.
+ InputStream socket =
+ new InputStream() {
+ @Override
+ public int read() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ reading.countDown();
+ awaitUninterruptibly(release);
+ return -1;
+ }
+
+ @Override
+ public void close() {
+ closing.countDown();
+ awaitUninterruptibly(release);
+ }
+ };
+ ReadableByteChannel channel = Channels.newChannel(socket);
+ List<String> reported = new CopyOnWriteArrayList<>();
+ GoogleStorageStallWatchdog interrupting =
+ new GoogleStorageStallWatchdog(
+ reported::add, () -> now, null,
GoogleStorageStallWatchdog.daemonInterrupter());
+
+ GoogleStorageStallWatchdog.Transfer stuck = interrupting.reading(URI,
LIMITS);
+ AtomicBoolean ended = new AtomicBoolean();
+ Thread reader =
+ new Thread(
+ () -> {
+ stuck.begin();
+ try {
+ channel.read(ByteBuffer.allocate(16));
+ } catch (IOException e) {
+ // Closed by the interrupt.
+ }
+ ended.set(stuck.end(0));
+ },
+ "stuck reader");
+ reader.setDaemon(true);
+ reader.start();
+ try {
+ assertTrue(reading.await(5, TimeUnit.SECONDS), "the read is under way");
+ advance(150);
+ interrupting.check();
+ assertTrue(closing.await(5, TimeUnit.SECONDS), "the interrupt is closing
the channel");
+
+ GoogleStorageStallWatchdog.Transfer other =
+ interrupting.reading("gs://bucket/other.txt", LIMITS);
+ other.begin();
+ advance(60);
+ assertTimeoutPreemptively(
+ Duration.ofSeconds(5), interrupting::check, "the checks wait for the
interrupt");
+ assertTrue(
+ reported.stream().anyMatch(line -> line.startsWith("No progress
reading gs://bucket/o")),
+ "the other transfer is reported: " + reported);
+ assertFalse(other.end(0));
+ } finally {
+ release.countDown();
+ }
+ reader.join(5000);
+ assertTrue(ended.get(), "the stuck read was ended by the watchdog");
+ }
+
+ @Test
+ void theDefaultLimits() {
+ GoogleStorageStallWatchdog.Limits limits =
+ GoogleStorageStallWatchdog.Limits.from(new GoogleCloudConfig());
+
+ assertEquals(Duration.ofSeconds(60), limits.readWarning(), "three read
timeouts, at least 60");
+ assertEquals(
+ Duration.ofSeconds(6 * (20 + 20) + 1 + 2 + 4 + 8 + 16),
+ limits.readAbort(),
+ "six attempts running into both timeouts, plus the delays between
them");
+ assertEquals(
+ Duration.ofSeconds(164), limits.writeWarning(), "a 16 MB chunk at 100
KB per second");
+ }
+
+ @Test
+ void theLimitsFollowTheTimeouts() {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ config.setReadTimeout("45");
+
+ GoogleStorageStallWatchdog.Limits limits =
GoogleStorageStallWatchdog.Limits.from(config);
+
+ assertEquals(Duration.ofSeconds(135), limits.readWarning());
+ assertEquals(Duration.ofSeconds(6 * (20 + 45) + 31), limits.readAbort());
+ assertEquals(Duration.ofSeconds(164), limits.writeWarning());
+ }
+
+ @Test
+ void theRetryDelaysAreCappedAtTheMaximumDelay() {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ config.setMaxAttempts("4");
+ config.setInitialRetryDelay("10");
+ config.setMaxRetryDelay("15");
+
+ assertEquals(
+ Duration.ofSeconds(4 * 40 + 10 + 15 + 15),
+ GoogleStorageStallWatchdog.Limits.from(config).readAbort());
+ }
+
+ @Test
+ void aReadIsNeverEndedBeforeItWasReported() {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ config.setMaxAttempts("1");
+
+ GoogleStorageStallWatchdog.Limits limits =
GoogleStorageStallWatchdog.Limits.from(config);
+
+ assertEquals(limits.readWarning(), limits.readAbort());
+ }
+
+ @Test
+ void theTotalTimeoutCapsTheAbort() {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ config.setTotalTimeout("2");
+ assertEquals(Duration.ofMinutes(2),
GoogleStorageStallWatchdog.Limits.from(config).readAbort());
+
+ config.setTotalTimeout("50");
+ config.setMaxAttempts("0");
+ assertEquals(
+ Duration.ofMinutes(50),
+ GoogleStorageStallWatchdog.Limits.from(config).readAbort(),
+ "unlimited attempts are bounded by the total timeout alone");
+ }
+
+ /** A cap below three read timeouts used to be raised back to the warning,
past the cap. */
+ @Test
+ void aTotalTimeoutBelowTheWarningEndsAndReportsTheReadInTime() {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ config.setReadTimeout("120");
+ config.setTotalTimeout("2");
+
+ GoogleStorageStallWatchdog.Limits limits =
GoogleStorageStallWatchdog.Limits.from(config);
+
+ assertEquals(Duration.ofMinutes(2), limits.readAbort(), "never later than
the total timeout");
+ assertEquals(Duration.ofMinutes(2), limits.readWarning(), "reported when
it is ended");
+ }
+
+ @Test
+ void noTimeoutCountsAsTheDefault() {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ config.setReadTimeout("");
+ config.setConnectionTimeout("-1");
+
+ assertEquals(
+ GoogleStorageStallWatchdog.Limits.from(new GoogleCloudConfig()),
+ GoogleStorageStallWatchdog.Limits.from(config));
+ }
+
+ /** The client takes a timeout of zero as no timeout: an attempt can then
take forever. */
+ @Test
+ void aZeroTimeoutLeavesOnlyTheTotalTimeout() {
+ for (String zero : List.of("read", "connect")) {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ if (zero.equals("read")) {
+ config.setReadTimeout("0");
+ } else {
+ config.setConnectionTimeout("0");
+ }
+
+ GoogleStorageStallWatchdog.Limits limits =
GoogleStorageStallWatchdog.Limits.from(config);
+
+ assertEquals(Duration.ofMinutes(50), limits.readAbort(), zero + "
timeout of zero");
+ assertEquals(Duration.ofSeconds(60), limits.readWarning(), zero + "
timeout of zero");
+ }
+ }
+
+ /** close() sends the last upload chunk while a write() can still be inside
the client. */
+ @Test
+ void overlappingCallsStayWatchedUntilTheLastOneIsDone() throws Exception {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.writing(URI,
LIMITS);
+ transfer.begin();
+ advance(30);
+
+ Thread closer = new Thread(transfer::begin);
+ closer.start();
+ closer.join();
+
+ // The write gives up; the close is still inside the client, and that
clock kept running.
+ transfer.failed();
+ advance(30);
+ watchdog.check();
+
+ assertEquals(1, logged.size(), "still watched after the first call ended:
" + logged);
+ assertTrue(logged.get(0).contains("for 60 seconds"), logged.get(0));
+ }
+
+ @Test
+ void aStalledReadThatFailsIsNotReportedAsResumed() {
+ GoogleStorageStallWatchdog.Transfer transfer = watchdog.reading(URI,
LIMITS);
+ transfer.begin();
+ advance(60);
+ watchdog.check();
+ advance(15);
+
+ assertFalse(transfer.failed());
+ assertEquals(1, logged.size(), "only the stall: " + logged);
+
+ advance(3600);
+ watchdog.check();
+ assertEquals(1, logged.size(), "and no longer watched");
+ }
+
+ private void advance(long seconds) {
+ now += TimeUnit.SECONDS.toNanos(seconds);
+ }
+
+ /** Wait the way a socket read does: an interrupt does not end it. Gives up
after 30 seconds. */
+ private static void awaitUninterruptibly(CountDownLatch latch) {
+ boolean interrupted = false;
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ while (latch.getCount() > 0 && System.nanoTime() < deadline) {
+ try {
+ latch.await(deadline - System.nanoTime(), TimeUnit.NANOSECONDS);
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+}
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageStalledDownloadTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageStalledDownloadTest.java
new file mode 100644
index 0000000000..23f235950a
--- /dev/null
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageStalledDownloadTest.java
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hop.vfs.gs;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.cloud.NoCredentials;
+import com.google.cloud.http.HttpTransportOptions;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hop.vfs.gs.config.GoogleCloudConfig;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.threeten.bp.Duration;
+
+/**
+ * Reproduces a Text File Input (or any VFS reader) hanging forever on a
{@code gs://} file.
+ *
+ * <p>The transform reads through {@link ReadChannelInputStream}, which wraps
{@code
+ * Storage.reader(...)}. In the JSON client that reader ends in {@code
+ * ApiaryUnbufferedReadableByteChannel.read}, a {@code do { ... } while
(true)} loop: when reading
+ * the response body fails with an exception the client considers retryable -
a {@code
+ * SocketTimeoutException} or {@code SocketException} - it drops the response
and opens the download
+ * again. Opening the download is bounded by the configured retry settings,
but the loop around the
+ * body read counts nothing, so neither the number of attempts nor the total
timeout is applied.
+ *
+ * <p>The result: an endpoint that answers with headers and then stops sending
the body keeps the
+ * reading thread busy forever, one read timeout per round. The same stall
before the headers fails
+ * after the configured attempts, which is what the user expects. On top of
that, the client's
+ * {@code close()} waits for the lock that the stuck {@code read()} holds, so
closing the stream
+ * from another thread cannot break the loop either.
+ *
+ * <p>Most tests here read through a stream the watchdog does not watch, to
pin what the client does
+ * on its own - the reason {@link GoogleStorageStallWatchdog} exists. If one
of them starts failing,
+ * the client may have been fixed. {@link
#theWatchdogReportsAndThenEndsAStalledDownload()} checks
+ * what Hop makes of the same stall.
+ */
+class GoogleStorageStalledDownloadTest {
+
+ private static final byte[] CONTENT =
+ "id;name\n1;one\n2;two\n3;three\n".getBytes(StandardCharsets.UTF_8);
+
+ /** Two attempts allowed: a well-behaved client gives up after the second
stalled download. */
+ private static final int MAX_ATTEMPTS = 2;
+
+ /** The configuration expresses this in whole seconds; milliseconds keep the
test fast. */
+ private static final int READ_TIMEOUT_MS = 250;
+
+ /** How many more downloads than allowed we wait for before calling the read
unbounded. */
+ private static final int EXTRA_DOWNLOADS = 4;
+
+ private enum Behaviour {
+ HEALTHY,
+ STALL_BEFORE_HEADERS,
+ STALL_AFTER_HEADERS
+ }
+
+ private HttpServer server;
+ private ExecutorService handlers;
+ private volatile Behaviour behaviour = Behaviour.HEALTHY;
+ private final AtomicInteger downloads = new AtomicInteger();
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final List<String> logged = new CopyOnWriteArrayList<>();
+ private final List<String> stalls = new CopyOnWriteArrayList<>();
+
+ /** Overrides the configured total timeout when set. */
+ private Duration totalTimeout;
+
+ /** Looks for stalls every 50 milliseconds, interrupting the way it does in
production. */
+ private final GoogleStorageStallWatchdog watchdog =
+ new GoogleStorageStallWatchdog(
+ stalls::add,
+ System::nanoTime,
+ java.time.Duration.ofMillis(50),
+ GoogleStorageStallWatchdog.daemonInterrupter());
+
+ @BeforeEach
+ void startEndpoint() throws IOException {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ // A stalled exchange parks its handler thread, so every request needs a
thread of its own or
+ // the next download would queue behind the stalled one and never reach
the endpoint.
+ handlers =
+ Executors.newCachedThreadPool(
+ r -> {
+ Thread t = new Thread(r, "fake-gcs-handler");
+ t.setDaemon(true);
+ return t;
+ });
+ server.setExecutor(handlers);
+ server.createContext("/", this::handle);
+ server.start();
+ }
+
+ @AfterEach
+ void stopEndpoint() {
+ watchdog.shutdown();
+ release.countDown();
+ server.stop(0);
+ handlers.shutdownNow();
+ }
+
+ private void handle(HttpExchange exchange) throws IOException {
+ downloads.incrementAndGet();
+ switch (behaviour) {
+ case HEALTHY:
+ objectHeaders(exchange);
+ exchange.sendResponseHeaders(200, CONTENT.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(CONTENT);
+ }
+ break;
+ case STALL_BEFORE_HEADERS:
+ awaitRelease();
+ exchange.close();
+ break;
+ case STALL_AFTER_HEADERS:
+ objectHeaders(exchange);
+ // The headers are flushed here, so the client sees a successful
open...
+ exchange.sendResponseHeaders(200, CONTENT.length);
+ // ...and then the body never arrives.
+ awaitRelease();
+ exchange.close();
+ break;
+ }
+ }
+
+ private static void objectHeaders(HttpExchange exchange) {
+ exchange.getResponseHeaders().add("Content-Type", "text/plain");
+ exchange.getResponseHeaders().add("x-goog-generation", "1");
+ exchange.getResponseHeaders().add("x-goog-metageneration", "1");
+ exchange.getResponseHeaders().add("x-goog-stored-content-length", "" +
CONTENT.length);
+ }
+
+ private void awaitRelease() {
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /** Sanity check of the fake endpoint: a normal object is read to the end
with one download. */
+ @Test
+ void aHealthyDownloadIsReadToTheEnd() throws Exception {
+ behaviour = Behaviour.HEALTHY;
+
+ try (InputStream in = openStream()) {
+ assertArrayEquals(CONTENT, readAll(in));
+ }
+ assertEquals(1, downloads.get());
+ }
+
+ /** Stalling before the headers is bounded: the read fails after the
configured attempts. */
+ @Test
+ void aStallBeforeTheHeadersGivesUpAfterTheConfiguredAttempts() {
+ behaviour = Behaviour.STALL_BEFORE_HEADERS;
+
+ assertThrows(
+ IOException.class,
+ () -> {
+ try (InputStream in = openStream()) {
+ readAll(in);
+ }
+ });
+ assertEquals(MAX_ATTEMPTS, downloads.get());
+ }
+
+ /**
+ * The reported hang. The same stall after the headers is retried without
any limit: the client
+ * opens the download over and over, far beyond the configured attempts, and
the read never
+ * returns or fails.
+ */
+ @Test
+ void aStallAfterTheHeadersIsRetriedForever() throws Exception {
+ behaviour = Behaviour.STALL_AFTER_HEADERS;
+
+ InputStream in = openStream();
+ StuckReader reader = StuckReader.start(in);
+
+ int expected = MAX_ATTEMPTS + EXTRA_DOWNLOADS;
+ assertTrue(
+ awaitDownloads(expected, 30_000),
+ "expected the client to keep re-opening the download, saw only " +
downloads.get());
+
+ assertTrue(
+ reader.thread.isAlive(),
+ "the read finished or failed, so the retry loop is bounded after all: "
+ + reader.outcome.get());
+ assertNull(reader.outcome.get(), "the read should neither return nor
fail");
+ assertTrue(
+ downloads.get() > MAX_ATTEMPTS,
+ downloads.get() + " downloads were made although only " + MAX_ATTEMPTS
+ " are allowed");
+
+ // The loop is no longer silent: every re-open follows a timeout that is
logged.
+ assertTrue(
+ logged.size() >= expected - 1,
+ "expected a logged timeout per re-open, got " + logged.size() + ": " +
logged);
+ assertTrue(
+ logged.get(0).contains("No response in time while accessing
gs://bucket/file.txt"),
+ logged.get(0));
+ }
+
+ /**
+ * What Hop makes of the same stall: the watchdog reports it, then ends it
once the limit is
+ * reached. The read fails with an error naming the file, the reading thread
is not left
+ * interrupted, and later reads fail too rather than pretend the file ended.
+ */
+ @Test
+ void theWatchdogReportsAndThenEndsAStalledDownload() throws Exception {
+ behaviour = Behaviour.STALL_AFTER_HEADERS;
+ InputStream in = openWatchedStream();
+
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ AtomicBoolean leftInterrupted = new AtomicBoolean(true);
+ AtomicReference<Throwable> laterRead = new AtomicReference<>();
+ Thread reader =
+ new Thread(
+ () -> {
+ try {
+ readAll(in);
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ leftInterrupted.set(Thread.currentThread().isInterrupted());
+ try {
+ in.read();
+ } catch (Throwable t) {
+ laterRead.set(t);
+ }
+ },
+ "gcs-watched-reader");
+ reader.setDaemon(true);
+ reader.start();
+ reader.join(15_000);
+
+ assertFalse(reader.isAlive(), "the watchdog did not end the stalled read");
+ assertInstanceOf(IOException.class, failure.get(), "the read should fail");
+ String message = failure.get().getMessage();
+ assertTrue(message.startsWith("No progress reading gs://bucket/file.txt
for "), message);
+ assertTrue(message.contains("giving up"), message);
+ assertFalse(leftInterrupted.get(), "the interrupt that ended the read
leaked out of it");
+ assertInstanceOf(
+ IOException.class, laterRead.get(), "a later read must fail too, not
report end of file");
+ assertTrue(
+ stalls.stream().anyMatch(line -> line.endsWith("still waiting.")),
+ "the stall should be reported before the read is ended: " + stalls);
+ }
+
+ /**
+ * The total timeout does not end it either. The client only looks at it
when an error is recorded
+ * while opening the download, against a budget that starts afresh with
every re-open; the failed
+ * body reads in between are never counted against it. Here it is one
second, and the read is
+ * still going after three.
+ */
+ @Test
+ void aStallAfterTheHeadersOutlivesTheTotalTimeout() throws Exception {
+ behaviour = Behaviour.STALL_AFTER_HEADERS;
+ totalTimeout = Duration.ofSeconds(1);
+
+ InputStream in = openStream();
+ StuckReader reader = StuckReader.start(in);
+ TimeUnit.SECONDS.sleep(3);
+
+ assertTrue(
+ reader.thread.isAlive(),
+ "the read ended, so the total timeout was applied after all: " +
reader.outcome.get());
+ assertNull(reader.outcome.get(), "the read should neither return nor
fail");
+ assertTrue(
+ downloads.get() > MAX_ATTEMPTS,
+ "expected the download to be re-opened beyond the configured attempts,
saw "
+ + downloads.get());
+ }
+
+ /**
+ * Closing the stream from another thread (a pipeline stop, say) does not
break the loop either:
+ * the client's close() waits for the lock its own stuck read() holds, so
the closing thread hangs
+ * as well.
+ */
+ @Test
+ void closingAStalledDownloadFromAnotherThreadBlocksToo() throws Exception {
+ behaviour = Behaviour.STALL_AFTER_HEADERS;
+
+ InputStream in = openStream();
+ StuckReader reader = StuckReader.start(in);
+ assertTrue(awaitDownloads(MAX_ATTEMPTS + 1, 30_000), "the read never got
going");
+
+ Thread closer = new Thread(StuckReader.quietly(in::close), "gcs-closer");
+ closer.setDaemon(true);
+ closer.start();
+ closer.join(10L * READ_TIMEOUT_MS);
+
+ assertTrue(closer.isAlive(), "close() returned while the read was stuck");
+ assertTrue(reader.thread.isAlive(), "close() broke the stuck read");
+ }
+
+ /** A stream the watchdog leaves alone: what the client does by itself. */
+ private InputStream openStream() {
+ return new ReadChannelInputStream(
+ storage().reader(BlobId.of("bucket", "file.txt")),
+ GoogleStorageStallWatchdog.Transfer.untracked("gs://bucket/file.txt"));
+ }
+
+ /** A watched stream, reported after four read timeouts and ended after
eight. */
+ private InputStream openWatchedStream() {
+ GoogleStorageStallWatchdog.Limits limits =
+ new GoogleStorageStallWatchdog.Limits(
+ java.time.Duration.ofMillis(4L * READ_TIMEOUT_MS),
+ java.time.Duration.ofMillis(8L * READ_TIMEOUT_MS),
+ java.time.Duration.ofMillis(4L * READ_TIMEOUT_MS));
+ return new ReadChannelInputStream(
+ storage().reader(BlobId.of("bucket", "file.txt")),
+ watchdog.reading("gs://bucket/file.txt", limits));
+ }
+
+ /**
+ * Builds the client the way {@link GoogleStorageFileSystem#setupStorage()}
does, pointed at the
+ * local endpoint. Retry delays are collapsed and the timeouts set in
milliseconds to keep the
+ * test fast; the attempt count - the budget the reader should respect - is
configured as a user
+ * would.
+ */
+ private Storage storage() {
+ GoogleCloudConfig config = new GoogleCloudConfig();
+ config.setMaxAttempts(Integer.toString(MAX_ATTEMPTS));
+
+ RetrySettings.Builder prompt =
+ GoogleStorageFileSystem.buildRetrySettings(config).toBuilder()
+ .setInitialRetryDelay(Duration.ofMillis(1))
+ .setMaxRetryDelay(Duration.ofMillis(2));
+ if (totalTimeout != null) {
+ prompt.setTotalTimeout(totalTimeout);
+ }
+
+ return GoogleStorageFileSystem.buildStorageOptions(config, logged::add)
+ .setRetrySettings(prompt.build())
+ .setTransportOptions(
+ HttpTransportOptions.newBuilder()
+ .setConnectTimeout(READ_TIMEOUT_MS)
+ .setReadTimeout(READ_TIMEOUT_MS)
+ .build())
+ .setHost("http://127.0.0.1:"; + server.getAddress().getPort())
+ .setProjectId("hop-test")
+ .setCredentials(NoCredentials.getInstance())
+ .build()
+ .getService();
+ }
+
+ private boolean awaitDownloads(int count, long timeoutMs) throws
InterruptedException {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (downloads.get() < count) {
+ if (System.currentTimeMillis() > deadline) {
+ return false;
+ }
+ TimeUnit.MILLISECONDS.sleep(20);
+ }
+ return true;
+ }
+
+ private static byte[] readAll(InputStream in) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ in.transferTo(out);
+ return out.toByteArray();
+ }
+
+ /** A transform thread reading the stream to the end, recording how the read
ended, if ever. */
+ private static final class StuckReader {
+ private final AtomicReference<String> outcome = new AtomicReference<>();
+ private Thread thread;
+
+ static StuckReader start(InputStream in) {
+ StuckReader reader = new StuckReader();
+ reader.thread =
+ new Thread(
+ () -> {
+ try {
+ reader.outcome.set("returned " + readAll(in).length + "
bytes");
+ } catch (Exception e) {
+ reader.outcome.set("failed with " + e);
+ }
+ },
+ "gcs-reader");
+ reader.thread.setDaemon(true);
+ reader.thread.start();
+ return reader;
+ }
+
+ static Runnable quietly(IoAction action) {
+ return () -> {
+ try {
+ action.run();
+ } catch (IOException ignored) {
+ // only whether close() returns matters here
+ }
+ };
+ }
+ }
+
+ @FunctionalInterface
+ private interface IoAction {
+ void run() throws IOException;
+ }
+}
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ReadChannelInputStreamTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ReadChannelInputStreamTest.java
index a2d302c616..d90577a958 100644
---
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ReadChannelInputStreamTest.java
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ReadChannelInputStreamTest.java
@@ -19,17 +19,25 @@
package org.apache.hop.vfs.gs;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.StorageException;
+import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
/**
@@ -43,9 +51,24 @@ import org.junit.jupiter.api.Test;
* {@code close()} aborts the in-flight read.
*
* <p>If these assertions ever fail again the monitor-deadlock has been
reintroduced.
+ *
+ * <p>This only covers Hop's own stream, against a channel that gives way on
close. The real GCS
+ * client does not: its {@code close()} waits for a read in progress (see
{@link
+ * GoogleStorageStalledDownloadTest}), so a stuck read is ended by {@link
+ * GoogleStorageStallWatchdog} instead.
*/
class ReadChannelInputStreamTest {
+ /** Report after a minute, end a read after two and a half. */
+ private static final GoogleStorageStallWatchdog.Limits LIMITS =
+ new GoogleStorageStallWatchdog.Limits(
+ Duration.ofSeconds(60), Duration.ofSeconds(150),
Duration.ofSeconds(60));
+
+ @AfterEach
+ void clearInterrupt() {
+ Thread.interrupted();
+ }
+
@Test
void closeBreaksAStuckReadAndDoesNotDeadlock() throws Exception {
CountDownLatch readEntered = new CountDownLatch(1);
@@ -107,4 +130,64 @@ class ReadChannelInputStreamTest {
assertTrue(
readReturned.get(), "close() did not unblock the stalled read() —
pipeline would freeze");
}
+
+ @Test
+ void aReadEndedByTheWatchdogFailsWithItsReason() throws Exception {
+ assertEndedReadFails(new ClosedByInterruptException());
+ }
+
+ @Test
+ void
aReadEndedByTheWatchdogFailsWithItsReasonWhenTheClientThrowsARuntimeException()
+ throws Exception {
+ assertEndedReadFails(new StorageException(0, "interrupted"));
+ }
+
+ /**
+ * The watchdog ends a stuck read by interrupting it. Whatever the client
then throws, the stream
+ * fails with the watchdog's reason, on that read and every later one, and
the interrupt is gone.
+ */
+ private static void assertEndedReadFails(Exception clientError) throws
Exception {
+ long[] now = {0};
+ GoogleStorageStallWatchdog watchdog =
+ new GoogleStorageStallWatchdog(message -> {}, () -> now[0], null,
Runnable::run);
+ ReadChannel stuck = mock(ReadChannel.class);
+ when(stuck.read(any(ByteBuffer.class)))
+ .thenAnswer(
+ inv -> {
+ now[0] += TimeUnit.SECONDS.toNanos(150);
+ watchdog.check(); // interrupts this thread, as the read is
stuck for too long
+ throw clientError;
+ });
+ ReadChannelInputStream in =
+ new ReadChannelInputStream(stuck,
watchdog.reading("gs://bucket/file.txt", LIMITS));
+
+ IOException error = assertThrows(IOException.class, () -> in.read(new
byte[16]));
+ assertTrue(error.getMessage().contains("giving up"), error.getMessage());
+ assertFalse(Thread.currentThread().isInterrupted(), "the interrupt must
not outlive the read");
+
+ IOException again = assertThrows(IOException.class, () -> in.read(new
byte[16]));
+ assertTrue(
+ again.getMessage().contains("giving up"), "not end-of-stream: the file
is cut short");
+ }
+
+ /** An Error from the client used to leave the transfer busy, and its thread
interruptible. */
+ @Test
+ void anErrorFromTheClientStillEndsTheTransfer() throws Exception {
+ List<String> logged = new ArrayList<>();
+ long[] now = {TimeUnit.HOURS.toNanos(1)};
+ GoogleStorageStallWatchdog watchdog =
+ new GoogleStorageStallWatchdog(logged::add, () -> now[0], null,
Runnable::run);
+
+ ReadChannel broken = mock(ReadChannel.class);
+ when(broken.read(any(ByteBuffer.class))).thenThrow(new
NoClassDefFoundError("io/grpc/Context"));
+ ReadChannelInputStream in =
+ new ReadChannelInputStream(broken,
watchdog.reading("gs://bucket/file.txt", LIMITS));
+
+ assertThrows(NoClassDefFoundError.class, () -> in.read(new byte[16], 0,
16));
+
+ now[0] += TimeUnit.HOURS.toNanos(1);
+ watchdog.check();
+ assertTrue(logged.isEmpty(), "no longer watched: " + logged);
+ assertFalse(Thread.currentThread().isInterrupted());
+ }
}
