Interface ExactlyOnceProducer

All Superinterfaces:
AutoCloseable, ProducerBase
@Experimental public interface ExactlyOnceProducer extends ProducerBase
Sends invocations to Restate with exactly-once deduplication. 
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());
}

Buffering

With a positive ProducerOptions.bufferMemory(), send(long, Invocation) 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 send instead waits until the protocol and transport are writable. Either wait is bounded by ProducerOptions.maxBlockTime(). The returned future tracks durable acknowledgement, not admission. Send several invocations without awaiting each future, then use ProducerBase.flush() or ProducerBase.flushAsync() to await them in bulk. ProducerBase.close() does not flush.

Non-blocking admission

For event-loop or callback-based code, trySend(long, Invocation) does not wait for admission. SendAttempt.Accepted contains the durable-acknowledgement future. On SendAttempt.Backpressured, use 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. 
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);
}

Producer identity and deduplication

Each invocation has a strictly increasing offset. Restate deduplicates on (producerId, offset), so choose a producer id that is stable across restarts and distinct per independent offset sequence: for example, a Kafka groupId/topic/partition or a Postgres logical-replication slot.

After a crash, replay from the last checkpoint; Restate drops already-committed offsets. ProducerBase.flush() and ProducerBase.waitAcknowledged(long) report the durable watermark to checkpoint, and ProducerBase.lastAcknowledgedOffset() remains available after a stream failure.

A producer is not thread-safe and fails fast with ConcurrentModificationException if used from more than one thread at once.