This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f64654e9749 add HTTP options to java-storage client
f64654e9749 is described below
commit f64654e97490963f2956d5a632e7a0303662be6a
Author: Ravi Dutt Singh <[email protected]>
AuthorDate: Wed Aug 16 19:20:57 2023 +0530
add HTTP options to java-storage client
---
docs/content.zh/docs/deployment/filesystems/gcs.md | 4 ++-
docs/content/docs/deployment/filesystems/gcs.md | 8 +++---
.../apache/flink/fs/gs/GSFileSystemFactory.java | 12 +++++++++
.../apache/flink/fs/gs/GSFileSystemOptions.java | 31 ++++++++++++++++++++++
4 files changed, 51 insertions(+), 4 deletions(-)
diff --git a/docs/content.zh/docs/deployment/filesystems/gcs.md
b/docs/content.zh/docs/deployment/filesystems/gcs.md
index ed889c5ec60..444c3d0d5a3 100644
--- a/docs/content.zh/docs/deployment/filesystems/gcs.md
+++ b/docs/content.zh/docs/deployment/filesystems/gcs.md
@@ -83,7 +83,9 @@ You can also set `gcs-connector` options directly in the
Hadoop `core-site.xml`
|---------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
| gs.writer.temporary.bucket.name | Set this property to choose a bucket to
hold temporary blobs for in-progress writes via `RecoverableWriter`. If this
property is not set, temporary blobs will be written to same bucket as the
final file being written. In either case, temporary blobs are written with the
prefix `.inprogress/`. <br><br> It is recommended to choose a separate bucket
in order to [assign it a TTL](https://cloud.google.com/storage/docs/lifecycle),
to provide a mechanism to [...]
| gs.writer.chunk.size | Set this property to [set the chunk
size](https://cloud.google.com/java/docs/reference/google-cloud-core/latest/com.google.cloud.WriteChannel#com_google_cloud_WriteChannel_setChunkSize_int_)
for writes via `RecoverableWriter`. <br><br>If not set, a Google-determined
default chunk size will be used.
[...]
-| gs.filesink.entropy.enabled | Set this property to improve performance
due to hotspotting issues on GCS. This option defines whether to enable entropy
injection in filesink gcs path. If this is enabled, entropy in the form of
temporary object id will be injected in beginning of the gcs path of the
temporary objects. The final object path remains unchanged.
[...]
+| gs.filesink.entropy.enabled | Set this property to improve performance
due to hotspotting issues on GCS. This option defines whether to enable entropy
injection in filesink gcs path. If this is enabled, entropy in the form of
temporary object id will be injected in beginning of the gcs path of the
temporary objects. The final object path remains unchanged.
[...]
+| gs.http.connect-timeout | Set this property to [set the connection
timeout](https://cloud.google.com/java/docs/reference/google-cloud-core/latest/com.google.cloud.http.HttpTransportOptions.Builder#com_google_cloud_http_HttpTransportOptions_Builder_setConnectTimeout_int_)
for java-storage client.
[...]
+| gs.http.read-timeout | Set this property to [set the content read
timeout](https://cloud.google.com/java/docs/reference/google-cloud-core/latest/com.google.cloud.http.HttpTransportOptions.Builder#com_google_cloud_http_HttpTransportOptions_Builder_setReadTimeout_int_)
from connection established via java-storage client.
[...]
### Authentication to access GCS
diff --git a/docs/content/docs/deployment/filesystems/gcs.md
b/docs/content/docs/deployment/filesystems/gcs.md
index 58670cb3355..8f0f69fafcb 100644
--- a/docs/content/docs/deployment/filesystems/gcs.md
+++ b/docs/content/docs/deployment/filesystems/gcs.md
@@ -79,11 +79,13 @@ You can also set `gcs-connector` options directly in the
Hadoop `core-site.xml`
`flink-gs-fs-hadoop` can also be configured by setting the following options
in [Flink configuration file]({{< ref
"docs/deployment/config#flink-configuration-file" >}}):
-| Key | Description
[...]
-|---------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| Key | Description
[...]
+|---------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
| gs.writer.temporary.bucket.name | Set this property to choose a bucket to
hold temporary blobs for in-progress writes via `RecoverableWriter`. If this
property is not set, temporary blobs will be written to same bucket as the
final file being written. In either case, temporary blobs are written with the
prefix `.inprogress/`. <br><br> It is recommended to choose a separate bucket
in order to [assign it a TTL](https://cloud.google.com/storage/docs/lifecycle),
to provide a mechanism to [...]
-| gs.writer.chunk.size | Set this property to [set the chunk
size](https://cloud.google.com/java/docs/reference/google-cloud-core/latest/com.google.cloud.WriteChannel#com_google_cloud_WriteChannel_setChunkSize_int_)
for writes via `RecoverableWriter`. <br><br>If not set, a Google-determined
default chunk size will be used.
[...]
+| gs.writer.chunk.size | Set this property to [set the chunk
size](https://cloud.google.com/java/docs/reference/google-cloud-core/latest/com.google.cloud.WriteChannel#com_google_cloud_WriteChannel_setChunkSize_int_)
for writes via `RecoverableWriter`. <br><br>If not set, a Google-determined
default chunk size will be used.
[...]
| gs.filesink.entropy.enabled | Set this property to improve performance
due to hotspotting issues on GCS. This option defines whether to enable entropy
injection in filesink gcs path. If this is enabled, entropy in the form of
temporary object id will be injected in beginning of the gcs path of the
temporary objects. The final object path remains unchanged.
[...]
+| gs.http.connect-timeout | Set this property to [set the connection
timeout](https://cloud.google.com/java/docs/reference/google-cloud-core/latest/com.google.cloud.http.HttpTransportOptions.Builder#com_google_cloud_http_HttpTransportOptions_Builder_setConnectTimeout_int_)
for java-storage client.
[...]
+| gs.http.read-timeout | Set this property to [set the content read
timeout](https://cloud.google.com/java/docs/reference/google-cloud-core/latest/com.google.cloud.http.HttpTransportOptions.Builder#com_google_cloud_http_HttpTransportOptions_Builder_setReadTimeout_int_)
from connection established via java-storage client.
[...]
### Authentication to access GCS
diff --git
a/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemFactory.java
b/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemFactory.java
index 0d4a337d8f6..79be5219743 100644
---
a/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemFactory.java
+++
b/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemFactory.java
@@ -27,6 +27,7 @@ import org.apache.flink.util.Preconditions;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem;
+import com.google.cloud.http.HttpTransportOptions;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import org.apache.hadoop.fs.Path;
@@ -92,6 +93,7 @@ public class GSFileSystemFactory implements FileSystemFactory
{
LOGGER.info("Using file system options {}", fileSystemOptions);
StorageOptions.Builder storageOptionsBuilder =
StorageOptions.newBuilder();
+
storageOptionsBuilder.setTransportOptions(getHttpTransportOptions(fileSystemOptions));
// get storage credentials
Optional<GoogleCredentials> credentials =
@@ -104,6 +106,16 @@ public class GSFileSystemFactory implements
FileSystemFactory {
this.storage = storageOptionsBuilder.build().getService();
}
+ private HttpTransportOptions getHttpTransportOptions(GSFileSystemOptions
fileSystemOptions) {
+ Optional<Integer> connectionTimeout =
fileSystemOptions.getHTTPConnectionTimeout();
+ Optional<Integer> readTimeout = fileSystemOptions.getHTTPReadTimeout();
+ HttpTransportOptions.Builder httpTransportOptionsBuilder =
+ HttpTransportOptions.newBuilder();
+
connectionTimeout.ifPresent(httpTransportOptionsBuilder::setConnectTimeout);
+ readTimeout.ifPresent(httpTransportOptionsBuilder::setReadTimeout);
+ return httpTransportOptionsBuilder.build();
+ }
+
@Override
public String getScheme() {
return SCHEME;
diff --git
a/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemOptions.java
b/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemOptions.java
index 71c0d7593e9..08402d96e7b 100644
---
a/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemOptions.java
+++
b/flink-filesystems/flink-gs-fs-hadoop/src/main/java/org/apache/flink/fs/gs/GSFileSystemOptions.java
@@ -59,6 +59,27 @@ public class GSFileSystemOptions {
+ "gcs path of the temporary objects. The
final object path "
+ "remains unchanged.");
+ /**
+ * Flink config option to set the http connection timeout. It will be used
by cloud-storage
+ * library.
+ */
+ public static final ConfigOption<Integer> GCS_HTTP_CONNECT_TIMEOUT =
+ ConfigOptions.key("gs.http.connect-timeout")
+ .intType()
+ .defaultValue(20_000)
+ .withDescription(
+ "This option sets the timeout in milliseconds to
establish a connection.");
+
+ /**
+ * Flink config option to set the http read timeout. It will be used by
cloud-storage library.
+ */
+ public static final ConfigOption<Integer> GCS_HTTP_READ_TIMEOUT =
+ ConfigOptions.key("gs.http.read-timeout")
+ .intType()
+ .defaultValue(20_000)
+ .withDescription(
+ "This option sets the timeout in milliseconds to
read data from an established connection.");
+
/** The Flink configuration. */
private final Configuration flinkConfig;
@@ -87,6 +108,16 @@ public class GSFileSystemOptions {
return flinkConfig.getOptional(WRITER_TEMPORARY_BUCKET_NAME);
}
+ /** Timeout in millisecond to establish the connection. */
+ public Optional<Integer> getHTTPConnectionTimeout() {
+ return flinkConfig.getOptional(GCS_HTTP_CONNECT_TIMEOUT);
+ }
+
+ /** Timeout in millisecond to read content from connection. */
+ public Optional<Integer> getHTTPReadTimeout() {
+ return flinkConfig.getOptional(GCS_HTTP_READ_TIMEOUT);
+ }
+
/** The chunk size to use for writes on the underlying Google
WriteChannel. */
public Optional<MemorySize> getWriterChunkSize() {
return flinkConfig.getOptional(WRITER_CHUNK_SIZE);
