Integration client

build.gradle.kts

@@ -68,6 +68,7 @@ val dokkaDocumentedProjects =
               "examples",
               "sdk-aggregated-javadocs",
               "admin-client",
+              "integration-client",
               "test-services",
           )
     }

gradle/libs.versions.toml

@@ -16,6 +16,30 @@
   bytebuddy = "net.bytebuddy:byte-buddy:1.18.3"
   objenesis = "org.objenesis:objenesis:3.4"
 
+  [libraries.grpc-protobuf]
+    module = 'io.grpc:grpc-protobuf'
+
+    [libraries.grpc-protobuf.version]
+      ref = 'grpc'
+
+  [libraries.grpc-stub]
+    module = 'io.grpc:grpc-stub'
+
+    [libraries.grpc-stub.version]
+      ref = 'grpc'
+
+  [libraries.grpc-netty-shaded]
+    module = 'io.grpc:grpc-netty-shaded'
+
+    [libraries.grpc-netty-shaded.version]
+      ref = 'grpc'
+
+  [libraries.grpc-inprocess]
+    module = 'io.grpc:grpc-inprocess'
+
+    [libraries.grpc-inprocess.version]
+      ref = 'grpc'
+
   [libraries.jackson-annotations]
     module = 'com.fasterxml.jackson.core:jackson-annotations'
     version = '2.22'
@@ -262,6 +286,7 @@
       ref = 'ksp'
 
 [versions]
+  grpc = '1.70.0'
   jackson = '2.22.0'
   junit = '5.14.1'
   kotlinx-coroutines = '1.10.2'

integration-client/build.gradle.kts

@@ -0,0 +1,119 @@
+import com.google.protobuf.gradle.id
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+plugins {
+  `java-library`
+  `java-conventions`
+  `library-publishing-conventions`
+  alias(libs.plugins.protobuf)
+}
+
+description = "Client for the Restate ingress integration (ingestion) API"
+
+dependencies {
+  // Generated protobuf messages + gRPC stubs are part of the public API surface.
+  api(libs.protobuf.java)
+  api(libs.grpc.protobuf)
+  api(libs.grpc.stub)
+
+  // Netty transport shipped as the default runtime channel implementation.
+  implementation(libs.grpc.netty.shaded)
+
+  // grpc-java generated stubs reference javax.annotation.Generated.
+  compileOnly(libs.tomcat.annotations)
+
+  // @ApiStatus.Experimental markers on the public API.
+  compileOnly(libs.jetbrains.annotations)
+  compileOnly(libs.jspecify)
+
+  testImplementation(libs.junit.jupiter)
+  testImplementation(libs.assertj)
+  testImplementation(libs.protobuf.java)
+  testImplementation(project(":sdk-api"))
+  testImplementation(project(":sdk-serde-jackson"))
+  testImplementation(project(":sdk-testing"))
+  // In-process transport to drive the client against a fake IngestionSvc in unit tests.
+  testImplementation(libs.grpc.inprocess)
+  testRuntimeOnly(libs.log4j.core)
+  testRuntimeOnly(libs.junit.platform.launcher)
+}
+
+// Code generation: protobuf messages + gRPC Java stubs for the ingestion service.
+protobuf {
+  protoc { artifact = "com.google.protobuf:protoc:${libs.versions.protobuf.get()}" }
+  plugins { id("grpc") { artifact = "io.grpc:protoc-gen-grpc-java:${libs.versions.grpc.get()}" } }
+  generateProtoTasks { all().forEach { it.plugins { id("grpc") } } }
+}
+
+// Generate a `Version` class at build time. VERSION comes from the Gradle `version`, GIT_HASH from
+// git; INTEGRATION is the default `name/version` identity stamped into the ingestion Start frame
+// when the caller doesn't provide its own via IntegrationClient.Builder.integration(...).
+val generatedVersionDir = layout.buildDirectory.dir("version")
+
+generatedVersionDir.get().asFile.mkdirs()
+
+// The protobuf plugin already registers the generated proto/grpc sources; only the version dir
+// needs wiring here.
+sourceSets { main { java { srcDir(generatedVersionDir) } } }
+
+// From https://discuss.kotlinlang.org/t/use-git-hash-as-version-number-in-build-gradle-kts/19818/4
+fun String.runCommand(
+    workingDir: File = File("."),
+    timeoutAmount: Long = 5,
+    timeoutUnit: TimeUnit = TimeUnit.SECONDS,
+): String =
+    ProcessBuilder(split("\\s(?=(?:[^'\"`]*(['\"`])[^'\"`]*\\1)*[^'\"`]*$)".toRegex()))
+        .directory(workingDir)
+        .redirectOutput(ProcessBuilder.Redirect.PIPE)
+        .redirectError(ProcessBuilder.Redirect.PIPE)
+        .start()
+        .apply { waitFor(timeoutAmount, timeoutUnit) }
+        .run {
+          val error = errorStream.bufferedReader().readText().trim()
+          if (error.isNotEmpty()) {
+            throw IllegalStateException(error)
+          }
+          inputStream.bufferedReader().readText().trim()
+        }
+
+val generateVersionClass =
+    tasks.register("generateVersionClass") {
+      dependsOn(project.tasks.processResources)
+      outputs.dir(generatedVersionDir)
+
+      doFirst {
+        // Tolerate a checkout with no commits yet (git rev-parse fails before the first commit).
+        val gitHash =
+            try {
+              "git rev-parse --short=8 HEAD".runCommand(workingDir = rootDir).ifBlank { "unknown" }
+            } catch (e: Exception) {
+              "unknown"
+            }
+        val containingDir = generatedVersionDir.get().dir("dev/restate/integration").asFile
+        assert(containingDir.exists() || containingDir.mkdirs())
+
+        file("$containingDir/Version.java")
+            .writeText(
+                """
+      package dev.restate.integration;
+
+      /** Generated at build time by the `generateVersionClass` Gradle task. Do not edit. */
+      public final class Version {
+          private Version() {}
+
+          public static final String VERSION = "$version";
+          public static final String GIT_HASH = "$gitHash";
+          // Default integration identifier for the ingestion Start frame: `name/version`.
+          public static final String INTEGRATION = "restate-integration-client/" + VERSION + "_" + GIT_HASH;
+      }
+      """
+                    .trimIndent()
+            )
+      }
+    }
+
+tasks {
+  withType<JavaCompile>().configureEach { dependsOn(generateVersionClass) }
+  withType<org.gradle.jvm.tasks.Jar>().configureEach { dependsOn(generateVersionClass) }
+}

integration-client/src/main/java/dev/restate/integration/ExactlyOnceProducer.java

@@ -0,0 +1,125 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.integration;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Sends invocations to Restate with exactly-once deduplication.
+ *
+ * <pre>{@code
+ * try (IntegrationClient client = IntegrationClient.builder("http://localhost:8080").build();
+ *     ExactlyOnceProducer producer =
+ *         client.newExactlyOnceProducer("group-a/orders/0")) {
+ *   long offset = checkpoint.load() + 1;
+ *   producer.send(
+ *       offset,
+ *       Invocation.create()
+ *           .setServiceName("Orders")
+ *           .setHandlerName("ingest")
+ *           .setBody(payload));
+ *   checkpoint.store(producer.flush());
+ * }
+ * }</pre>
+ *
+ * <h2>Buffering</h2>
+ *
+ * With a positive {@link ProducerOptions#bufferMemory()}, {@link #send} first admits the invocation
+ * to a byte-bounded local buffer while it waits to be handed to the transport. A value of zero
+ * disables local buffering, so {@code send} instead waits until the protocol and transport are
+ * writable. Either wait is bounded by {@link ProducerOptions#maxBlockTime()}. The returned future
+ * tracks durable acknowledgement, not admission. Send several invocations without awaiting each
+ * future, then use {@link #flush()} or {@link #flushAsync()} to await them in bulk. {@link
+ * #close()} does not flush.
+ *
+ * <h2>Non-blocking admission</h2>
+ *
+ * For event-loop or callback-based code, {@link #trySend} does not wait for admission. {@link
+ * SendAttempt.Accepted} contains the durable-acknowledgement future. On {@link
+ * SendAttempt.Backpressured}, use {@link SendAttempt.Backpressured#ready()} to schedule a retry of
+ * the same offset and invocation on the event loop; readiness is a notification, not an admission
+ * reservation.
+ *
+ * <pre>{@code
+ * static CompletableFuture<SendResult> sendWithoutBlocking(
+ *     ExactlyOnceProducer producer, long offset, Invocation invocation, Executor eventLoop) {
+ *   SendAttempt attempt = producer.trySend(offset, invocation);
+ *   if (attempt instanceof SendAttempt.Accepted accepted) {
+ *     return accepted.acknowledgement();
+ *   }
+ *   return ((SendAttempt.Backpressured) attempt)
+ *       .ready()
+ *       .thenComposeAsync(
+ *           ignored -> sendWithoutBlocking(producer, offset, invocation, eventLoop), eventLoop);
+ * }
+ * }</pre>
+ *
+ * <h2>Producer identity and deduplication</h2>
+ *
+ * Each invocation has a strictly increasing offset. Restate deduplicates on {@code (producerId,
+ * offset)}, so choose a producer id that is <b>stable across restarts</b> and <b>distinct per
+ * independent offset sequence</b>: for example, a Kafka {@code groupId/topic/partition} or a
+ * Postgres logical-replication slot.
+ *
+ * <p>After a crash, replay from the last checkpoint; Restate drops already-committed offsets.
+ * {@link #flush()} and {@link #waitAcknowledged(long)} report the durable watermark to checkpoint,
+ * and {@link #lastAcknowledgedOffset()} remains available after a stream failure.
+ *
+ * <p>A producer is <b>not thread-safe</b> and fails fast with {@link
+ * java.util.ConcurrentModificationException} if used from more than one thread at once.
+ */
+@org.jetbrains.annotations.ApiStatus.Experimental
+public interface ExactlyOnceProducer extends ProducerBase {
+
+  /**
+   * Sends an invocation at {@code offset}.
+   *
+   * <p>This method waits up to {@link ProducerOptions#maxBlockTime()} when the local buffer is
+   * full, or, when buffering is disabled, until protocol and transport readiness permit a direct
+   * write. The invocation is refused with {@link ProducerBufferExhaustedException} if the timeout
+   * elapses. A zero duration makes this method fail immediately under backpressure.
+   *
+   * <p>The returned future completes when the invocation is durably committed by Restate.
+   *
+   * @param offset the offset to assign to this record; must be strictly greater than the previous
+   *     one
+   * @param invocation the invocation to send
+   * @return a future completing, once the record is durably committed by Restate, with the {@link
+   *     SendResult} carrying {@code offset}
+   * @throws ProducerBufferExhaustedException if the producer cannot admit the invocation before the
+   *     configured maximum blocking time elapses, or the thread is interrupted while waiting
+   * @throws IllegalStateException if a reentrant producer callback invokes this method when it
+   *     would block
+   * @throws IllegalArgumentException if {@code offset} is not strictly greater than {@link
+   *     #lastSentOffset()}, or buffering is enabled and the serialized invocation is larger than
+   *     {@link ProducerOptions#bufferMemory()}
+   * @throws java.util.ConcurrentModificationException if the producer is used concurrently from
+   *     another thread
+   */
+  CompletableFuture<SendResult> send(long offset, Invocation invocation)
+      throws ProducerBufferExhaustedException;
+
+  /**
+   * Attempts to send an invocation at {@code offset} without blocking.
+   *
+   * <p>An {@link SendAttempt.Accepted} carries the durable-acknowledgement future. A {@link
+   * SendAttempt.Backpressured} carries a future that completes when retrying may succeed; the
+   * notification does not reserve admission.
+   *
+   * @param offset the offset to assign; must be strictly greater than the previous accepted offset
+   * @param invocation the invocation to send
+   * @return the admission result
+   * @throws IllegalArgumentException if {@code offset} is not strictly greater than {@link
+   *     #lastSentOffset()}, or buffering is enabled and the serialized invocation is larger than
+   *     {@link ProducerOptions#bufferMemory()}
+   * @throws java.util.ConcurrentModificationException if the producer is used concurrently from
+   *     another thread
+   */
+  SendAttempt trySend(long offset, Invocation invocation);
+}

integration-client/src/main/java/dev/restate/integration/GrpcIntegrationClient.java

@@ -0,0 +1,30 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.integration;
+
+import io.grpc.Channel;
+import org.jetbrains.annotations.ApiStatus;
+
+/**
+ * Internal bridge for first-party integrations that supply their own gRPC {@link Channel}.
+ *
+ * <p>The supplied channel remains owned by the caller and is not shut down when the client is
+ * closed.
+ *
+ * @hidden
+ */
+@ApiStatus.Internal
+public final class GrpcIntegrationClient {
+
+  private GrpcIntegrationClient() {}
+
+  public static IntegrationClient.Builder builder(Channel channel) {
+    return new IntegrationClient.Builder(channel);
+  }
+}

integration-client/src/main/java/dev/restate/integration/IngressEndpoint.java

@@ -0,0 +1,61 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.integration;
+
+import java.net.URI;
+
+/** Where to reach the Restate ingestion gRPC endpoint, parsed from an http(s) URL. */
+final class IngressEndpoint {
+
+  final String host;
+  final int port;
+  final boolean tls;
+
+  private IngressEndpoint(String host, int port, boolean tls) {
+    this.host = host;
+    this.port = port;
+    this.tls = tls;
+  }
+
+  /**
+   * Parse a single ingress URL. {@code https} selects TLS; the port defaults to 443 (TLS) or 80
+   * otherwise.
+   */
+  static IngressEndpoint parse(String raw) {
+    if (raw == null || raw.isBlank()) {
+      throw new IllegalArgumentException("ingress url must not be empty");
+    }
+    URI uri;
+    try {
+      uri = new URI(raw.trim());
+    } catch (Exception e) {
+      throw new IllegalArgumentException("ingress url is not a valid URL: '" + raw + "'", e);
+    }
+    boolean tls;
+    String scheme = uri.getScheme() == null ? null : uri.getScheme().toLowerCase();
+    if ("https".equals(scheme)) {
+      tls = true;
+    } else if ("http".equals(scheme)) {
+      tls = false;
+    } else {
+      throw new IllegalArgumentException(
+          "ingress url must use http or https scheme, got '"
+              + uri.getScheme()
+              + "' in '"
+              + raw
+              + "'");
+    }
+    String host = uri.getHost();
+    if (host == null) {
+      throw new IllegalArgumentException("ingress url has no host: '" + raw + "'");
+    }
+    int port = uri.getPort() != -1 ? uri.getPort() : (tls ? 443 : 80);
+    return new IngressEndpoint(host, port, tls);
+  }
+}