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

@@ -22,6 +22,30 @@
     [libraries.aws-sdk-kms.version]
       ref = 'aws-sdk'
 
+  [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'
@@ -275,6 +299,7 @@
 
 [versions]
   aws-sdk = '2.55.16'
+  grpc = '1.70.0'
   jackson = '2.22.0'
   junit = '5.14.1'
   kotlinx-coroutines = '1.10.2'

integration-client/README.md

@@ -0,0 +1,87 @@
+# Restate Integration Client
+
+You can use the Integration API to turn any source of events into Restate invocations.
+
+Use an integration client to send invocations to Restate using the Integration API.
+
+## Installation
+
+Add [`dev.restate:integration-client`](https://central.sonatype.com/artifact/dev.restate/integration-client) to your dependencies.
+
+## Create a client
+
+Create one client for your Restate ingress URL and close it when your integration shuts down:
+
+```java
+try (IntegrationClient client = IntegrationClient.builder("http://localhost:8080").build()) {
+  // Use client
+}
+```
+
+## Simple producer
+
+Use `Producer` when your source does not provide stable offsets that you can replay.
+The client assigns an offset to each invocation.
+Call `flush()` before closing the producer to wait until Restate has durably committed everything you sent:
+
+```java
+try (Producer producer = client.newProducer()) {
+  for (SourceRecord record : records) {
+    producer.send(
+        Invocation.create()
+            .setServiceName("OrderService")
+            .setHandlerName("ingest")
+            .putHeader("content-type", "application/json")
+            .setBody(record.payload()));
+  }
+  producer.flush();
+}
+```
+
+## Exactly once producer
+
+Use `ExactlyOnceProducer` when your source provides stable, strictly increasing offsets,
+and choose a producer ID that remains stable across restarts and is unique to that offset sequence.
+After `flush()` returns, you can checkpoint the returned offset in your source:
+
+```java
+try (ExactlyOnceProducer producer = client.newExactlyOnceProducer("orders/partition-0")) {
+  for (SourceRecord record : sourceRecords) {
+    producer.send(
+        record.offset(),
+        Invocation.create()
+            .setServiceName("OrderService")
+            .setHandlerName("ingest")
+            .putHeader("content-type", "application/json")
+            .setBody(record.payload()));
+  }
+  long committedOffset = producer.flush();
+  // Checkpoint committedOffset in your source
+}
+```
+
+## Producer default fields
+
+Producers accept `InvocationMetadata` when every invocation shares fields such as the target service, handler, or headers.
+Set defaults to avoid sending over and over the same fields in every invocation.
+
+Fields set on an individual invocation override these defaults.
+
+```java
+InvocationMetadata defaults =
+    InvocationMetadata.create()
+        .setServiceName("OrderService")
+        .setHandlerName("ingest")
+        .putHeader("content-type", "application/json");
+
+try (Producer producer = client.newProducer(defaults)) {
+  for (SourceRecord record : records) {
+    producer.send(Invocation.create().setBody(record.payload()));
+  }
+  producer.flush();
+}
+```
+
+## More
+
+For configuration options and advanced usage, see the [Integration client Javadocs](https://restatedev.github.io/sdk-java/javadocs/dev/restate/integration/package-summary.html).

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);
+}