Add a GCS external payload storage driver

CHANGELOG.md

@@ -25,6 +25,9 @@ to docs, or any other relevant information.
   `WorkflowOutboundCallsInterceptorBase` need to add them, and Base subclasses override them to see calls with options.
 
 ### Added
+- Added experimental Google Cloud Storage external payload storage through the optional
+  `temporal-payload-storage-gcsdriver` and `temporal-payload-storage-gcsdriver-gcssdk` modules,
+  with conditional creation, SHA-256 integrity verification, and cancellable executor-backed I/O.
 - Applications can use the experimental `@TransferTypeConvertible` annotation and
   `TransferTypeConverter` interface to convert top-level model values to transfer types. The SDK
   applies this conversion to workflow, activity, Nexus, and schedule payloads, including failure

contrib/temporal-payload-storage-gcsdriver-gcssdk/build.gradle

@@ -0,0 +1,22 @@
+description = '''Temporal Java SDK External Storage GCS Driver - Google Cloud Storage Client'''
+
+apply from: "$rootDir/gradle/fakeGcsServer.gradle"
+
+ext {
+    googleCloudStorageVersion = '2.75.0'
+}
+
+dependencies {
+    api project(':temporal-payload-storage-gcsdriver')
+    api "com.google.cloud:google-cloud-storage:$googleCloudStorageVersion"
+
+    // For the @Experimental annotation only.
+    compileOnly project(':temporal-sdk')
+
+    // Exercise the Google and Temporal dependencies together on the ordinary test classpath.
+    testImplementation project(':temporal-serviceclient')
+    testImplementation project(':temporal-sdk')
+    testImplementation "junit:junit:${junitVersion}"
+    testImplementation "org.mockito:mockito-core:${mockitoVersion}"
+    testRuntimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}"
+}

contrib/temporal-payload-storage-gcsdriver-gcssdk/src/main/java/io/temporal/payload/storage/gcsdriver/gcssdk/GcsStorageClientAdapter.java

@@ -0,0 +1,107 @@
+package io.temporal.payload.storage.gcsdriver.gcssdk;
+
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import io.temporal.common.Experimental;
+import io.temporal.payload.storage.gcsdriver.GcsStorageDriverClient;
+import java.net.HttpURLConnection;
+import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.function.Supplier;
+import javax.annotation.Nonnull;
+
+/**
+ * Google Cloud Storage adapter that runs blocking calls on an application-supplied executor.
+ *
+ * <p>The application configures credentials, transport, retries, and time limits on {@link
+ * Storage}. It owns both resources and must shut them down after Temporal's storage work finishes.
+ * Use a dedicated executor: {@code Executors.newVirtualThreadPerTaskExecutor()} on Java 21+, or a
+ * fixed thread pool sized to at least the worker's activity and workflow task slots on earlier
+ * versions.
+ *
+ * <p>Provider failures propagate unchanged. The storage driver adds operation and location context.
+ *
+ * <p>Cancellation interrupts the executor task on a best-effort basis. Configure Google client time
+ * limits to bound operations that do not respond to interruption. Cancellation cannot undo an
+ * upload.
+ */
+@Experimental
+public final class GcsStorageClientAdapter implements GcsStorageDriverClient {
+  private final Storage storage;
+  private final ExecutorService executor;
+
+  public GcsStorageClientAdapter(@Nonnull Storage storage, @Nonnull ExecutorService executor) {
+    this.storage = Objects.requireNonNull(storage, "storage");
+    this.executor = Objects.requireNonNull(executor, "executor");
+  }
+
+  @Nonnull
+  @Override
+  public CompletableFuture<Void> putObjectIfAbsent(
+      @Nonnull String bucket, @Nonnull String objectName, @Nonnull byte[] data) {
+    return submit(
+        () -> {
+          try {
+            storage.create(
+                BlobInfo.newBuilder(bucket, objectName)
+                    .setContentType("application/octet-stream")
+                    .build(),
+                data,
+                Storage.BlobTargetOption.doesNotExist());
+          } catch (StorageException e) {
+            if (e.getCode() != HttpURLConnection.HTTP_PRECON_FAILED) {
+              throw e;
+            }
+            // The doesNotExist() precondition failed because the object already exists.
+          }
+          return null;
+        });
+  }
+
+  @Nonnull
+  @Override
+  public CompletableFuture<byte[]> getObject(@Nonnull String bucket, @Nonnull String objectName) {
+    return submit(() -> storage.readAllBytes(BlobId.of(bucket, objectName)));
+  }
+
+  private <T> CompletableFuture<T> submit(Supplier<T> call) {
+    CompletableFuture<T> result = new CompletableFuture<>();
+    FutureTask<T> task =
+        new FutureTask<T>(call::get) {
+          @Override
+          protected void done() {
+            try {
+              result.complete(get());
+            } catch (CancellationException e) {
+              result.cancel(true);
+            } catch (ExecutionException e) {
+              result.completeExceptionally(e.getCause());
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+              result.completeExceptionally(e);
+            }
+          }
+        };
+    result.whenComplete(
+        (value, error) -> {
+          if (result.isCancelled()) {
+            task.cancel(true);
+          }
+        });
+    try {
+      // Wire cancellation before submission so even queued tasks can be cancelled safely.
+      executor.execute(task);
+    } catch (RejectedExecutionException e) {
+      result.completeExceptionally(e);
+      task.cancel(true);
+    }
+    return result;
+  }
+}

contrib/temporal-payload-storage-gcsdriver-gcssdk/src/test/java/io/temporal/payload/storage/gcsdriver/gcssdk/FakeGcsServer.java

@@ -0,0 +1,100 @@
+package io.temporal.payload.storage.gcsdriver.gcssdk;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import org.junit.rules.ExternalResource;
+
+final class FakeGcsServer extends ExternalResource {
+  private Process process;
+  private String endpoint;
+
+  String endpoint() {
+    return endpoint;
+  }
+
+  @Override
+  protected void before() throws Throwable {
+    int port;
+    try (ServerSocket socket = new ServerSocket(0, 0, InetAddress.getByName("127.0.0.1"))) {
+      port = socket.getLocalPort();
+    }
+    endpoint = "http://127.0.0.1:" + port;
+    Path log = Files.createTempFile("fake-gcs-server-", ".log");
+    log.toFile().deleteOnExit();
+    process =
+        new ProcessBuilder(
+                Objects.requireNonNull(
+                    System.getProperty("io.temporal.testing.fakeGcsServer"),
+                    "Run tests through Gradle to prepare fake-gcs-server"),
+                "-scheme",
+                "http",
+                "-host",
+                "127.0.0.1",
+                "-port",
+                Integer.toString(port),
+                "-backend",
+                "memory",
+                "-external-url",
+                endpoint)
+            .redirectErrorStream(true)
+            .redirectOutput(log.toFile())
+            .start();
+    try {
+      long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+      while (process.isAlive() && System.nanoTime() < deadline) {
+        if (isReady()) {
+          return;
+        }
+        Thread.sleep(50);
+      }
+      throw new IOException(
+          "Fake GCS server failed to start:\n"
+              + new String(Files.readAllBytes(log), StandardCharsets.UTF_8));
+    } catch (Throwable error) {
+      after();
+      throw error;
+    }
+  }
+
+  private boolean isReady() throws IOException {
+    HttpURLConnection connection =
+        (HttpURLConnection)
+            new URL(endpoint + "/storage/v1/b?project=temporal-test").openConnection();
+    connection.setConnectTimeout(1000);
+    connection.setReadTimeout(1000);
+    try {
+      return connection.getResponseCode() == 200;
+    } catch (IOException e) {
+      return false;
+    } finally {
+      connection.disconnect();
+    }
+  }
+
+  @Override
+  protected void after() {
+    if (process == null) {
+      return;
+    }
+    process.destroy();
+    try {
+      if (!process.waitFor(5, TimeUnit.SECONDS)) {
+        process.destroyForcibly();
+        if (!process.waitFor(5, TimeUnit.SECONDS)) {
+          throw new AssertionError("Fake GCS server did not stop");
+        }
+      }
+    } catch (InterruptedException e) {
+      process.destroyForcibly();
+      Thread.currentThread().interrupt();
+    }
+  }
+}