[Java] Add ingestRecordNoWait fire-and-forget ingestion API

java/NEXT_CHANGELOG.md

@@ -20,3 +20,5 @@
 ### Deprecations
 
 ### API Changes
+
+- Added `ingestRecordNoWait(...)` fire-and-forget ingestion APIs for Java proto and JSON streams, plus `ingestRecordsNoWait(...)` batch APIs. These methods return after handing payloads to the native background runtime and do not return offsets or surface background ingestion errors.

java/README.md

@@ -913,23 +913,43 @@ Stream for Protocol Buffer ingestion with method-level generics. Use `ZerobusSdk
 ```
 Ingests a Protocol Buffer message and returns the offset immediately.
 
+```java
+<T extends Message> void ingestRecordNoWait(T record) throws ZerobusException
+```
+Hands a protobuf message to the native runtime without returning an offset or waiting for acknowledgment.
+
 ```java
 long ingestRecordOffset(byte[] encodedBytes) throws ZerobusException
 ```
 Ingests pre-encoded bytes and returns the offset immediately.
 
+```java
+void ingestRecordNoWait(byte[] encodedBytes) throws ZerobusException
+```
+Hands pre-encoded bytes to the native runtime without returning an offset or waiting for acknowledgment.
+
 **Batch Methods:**
 
 ```java
 <T extends Message> Optional<Long> ingestRecordsOffset(Iterable<T> records) throws ZerobusException
 ```
 Ingests multiple messages and returns the batch offset.
 
+```java
+<T extends Message> void ingestRecordsNoWait(Iterable<T> records) throws ZerobusException
+```
+Hands multiple messages to the native runtime without returning an offset or waiting for acknowledgment.
+
 ```java
 Optional<Long> ingestRecordsOffset(List<byte[]> encodedRecords) throws ZerobusException
 ```
 Ingests multiple pre-encoded byte arrays and returns the batch offset.
 
+```java
+void ingestRecordsNoWait(List<byte[]> encodedRecords) throws ZerobusException
+```
+Hands multiple pre-encoded byte arrays to the native runtime without returning an offset or waiting for acknowledgment.
+
 **Recovery Methods:**
 
 ```java
@@ -962,23 +982,50 @@ Stream for JSON ingestion with method-level generics. Use `ZerobusSdk.createJson
 ```
 Ingests an object serialized to JSON and returns the offset immediately.
 
+```java
+<T> void ingestRecordNoWait(T object, JsonSerializer<T> serializer) throws ZerobusException
+```
+Hands an object serialized to JSON to the native runtime without returning an offset or waiting for acknowledgment.
+
 ```java
 long ingestRecordOffset(String json) throws ZerobusException
 ```
 Ingests a JSON string and returns the offset immediately.
 
+```java
+void ingestRecordNoWait(String json) throws ZerobusException
+```
+Hands a JSON string to the native runtime without returning an offset or waiting for acknowledgment.
+
 **Batch Methods:**
 
 ```java
 <T> Optional<Long> ingestRecordsOffset(Iterable<T> objects, JsonSerializer<T> serializer) throws ZerobusException
 ```
 Ingests multiple objects as JSON and returns the batch offset.
 
+```java
+<T> void ingestRecordsNoWait(Iterable<T> objects, JsonSerializer<T> serializer) throws ZerobusException
+```
+Hands multiple objects as JSON to the native runtime without returning an offset or waiting for acknowledgment.
+
 ```java
 Optional<Long> ingestRecordsOffset(Iterable<String> jsonStrings) throws ZerobusException
 ```
 Ingests multiple JSON strings and returns the batch offset.
 
+```java
+void ingestRecordsNoWait(Iterable<String> jsonStrings) throws ZerobusException
+```
+Hands multiple JSON strings to the native runtime without returning an offset or waiting for acknowledgment.
+
+**No-wait semantics:** no-wait methods return after synchronous validation and hand-off to a
+native background task. They do not wait for local enqueue, native backpressure, offset assignment,
+or server acknowledgment. Background ingestion errors are intentionally not surfaced to the caller,
+and no ordering is guaranteed relative to later offset-returning, `flush()`, `close()`, or
+`waitForOffset()` calls. Use `ingestRecordOffset()` / `ingestRecordsOffset()` when the caller needs
+an offset or deterministic inclusion in a later wait.
+
 **Recovery Methods:**
 
 ```java
@@ -1229,16 +1276,18 @@ Called when an error occurs for records at or after `offsetId`.
 
 1. **Reuse SDK instances**: Create one `ZerobusSdk` instance per application
 2. **Stream lifecycle**: Always close streams in a `finally` block or use try-with-resources
-3. **Use offset-based API for high throughput**: `ingestRecordOffset()` avoids `CompletableFuture` overhead
-4. **Batch records when possible**: Use `ingestRecordsOffset()` for multiple records
-5. **Configure `maxInflightRecords`**: Adjust based on your throughput and memory requirements
-6. **Implement proper error handling**: Distinguish between retriable and non-retriable errors
-7. **Use `AckCallback` for monitoring**: Track acknowledgment progress without blocking
-8. **Proto generation**: Use the built-in `GenerateProto` tool to generate proto files from table schemas
-9. **Choose the right API**:
+3. **Use no-wait ingestion when per-record offsets are unnecessary**: `ingestRecordNoWait()` returns after handing the payload to a native background task; it does not wait for local enqueue, offset assignment, or acknowledgment, and background ingestion errors are not surfaced to the caller
+4. **Use offset-based API for tracked ingestion**: `ingestRecordOffset()` avoids `CompletableFuture` overhead while still letting you wait on specific offsets
+5. **Batch records when possible**: Use `ingestRecordsOffset()` for multiple records
+6. **Configure `maxInflightRecords`**: Adjust based on your throughput and memory requirements
+7. **Implement proper error handling**: Distinguish between retriable and non-retriable errors
+8. **Use `AckCallback` for monitoring**: Track acknowledgment progress without blocking
+9. **Proto generation**: Use the built-in `GenerateProto` tool to generate proto files from table schemas
+10. **Choose the right API**:
    - `ingestRecord()` → Simple use cases, moderate throughput (deprecated)
+   - `ingestRecordNoWait()` → Fire-and-forget when per-record tracking is unnecessary
    - `ingestRecordOffset()` + `waitForOffset()` → High throughput, fine-grained control (recommended)
-10. **Recovery pattern**: Use `sdk.recreateStream(closedStream)` to automatically re-ingest unacknowledged records, or manually use `getUnackedBatches()` after stream close
+11. **Recovery pattern**: Use `sdk.recreateStream(closedStream)` to automatically re-ingest unacknowledged records, or manually use `getUnackedBatches()` after stream close
 
 ## Community and Contributing
 

java/examples/README.md

@@ -61,8 +61,12 @@ ZerobusProtoStream stream = sdk.createProtoStream(
 // Method-level generics - flexible typing
 stream.ingestRecordOffset(myProtoMessage);        // Message
 stream.ingestRecordOffset(preEncodedBytes);       // byte[]
+stream.ingestRecordNoWait(myProtoMessage);        // fire-and-forget
+stream.ingestRecordNoWait(preEncodedBytes);       // fire-and-forget bytes
 stream.ingestRecordsOffset(listOfMessages);       // batch
 stream.ingestRecordsOffset(listOfByteArrays);     // batch
+stream.ingestRecordsNoWait(listOfMessages);       // fire-and-forget batch
+stream.ingestRecordsNoWait(listOfByteArrays);     // fire-and-forget bytes batch
 ```
 
 ### ZerobusJsonStream (Recommended for JSON)
@@ -77,8 +81,12 @@ ZerobusJsonStream stream = sdk.createJsonStream(
 // Method-level generics - flexible typing
 stream.ingestRecordOffset(object, gson::toJson);  // Object + serializer
 stream.ingestRecordOffset(jsonString);            // String
+stream.ingestRecordNoWait(object, gson::toJson);  // fire-and-forget object
+stream.ingestRecordNoWait(jsonString);            // fire-and-forget String
 stream.ingestRecordsOffset(objects, gson::toJson);// batch
 stream.ingestRecordsOffset(jsonStrings);          // batch
+stream.ingestRecordsNoWait(objects, gson::toJson);// fire-and-forget batch
+stream.ingestRecordsNoWait(jsonStrings);          // fire-and-forget String batch
 ```
 
 ### ZerobusArrowStream (Experimental - Arrow Flight)

java/src/main/java/com/databricks/zerobus/BaseZerobusStream.java

@@ -194,6 +194,11 @@ protected void ensureOpen() throws ZerobusException {
   protected native CompletableFuture<Void> nativeIngestRecord(
       long handle, byte[] payload, boolean isJson);
 
+  protected native void nativeIngestRecordNoWait(long handle, byte[] payload, boolean isJson);
+
+  protected native void nativeIngestRecordsNoWait(
+      long handle, List<byte[]> payloads, boolean isJson);
+
   protected native long nativeIngestRecordOffset(long handle, byte[] payload, boolean isJson);
 
   protected native long nativeIngestRecordsOffset(

java/src/main/java/com/databricks/zerobus/ZerobusJsonStream.java

@@ -157,6 +157,28 @@ public <T> long ingestRecordOffset(T object, JsonSerializer<T> serializer)
     return nativeIngestRecordOffset(nativeHandle, json.getBytes(StandardCharsets.UTF_8), true);
   }
 
+  /**
+   * Ingests an object as JSON asynchronously without exposing the offset or ack future.
+   *
+   * <p>This fire-and-forget method serializes the object on the calling thread, then hands the
+   * encoded payload off to the native runtime. Enqueueing into the stream, native backpressure, and
+   * offset assignment happen later on the runtime's thread pool. Errors raised after this hand-off
+   * are intentionally not surfaced to the caller, and there is no ordering guarantee relative to
+   * subsequent offset-returning, {@link #flush()}, {@link #close()}, or {@link #waitForOffset(long)}
+   * calls.
+   *
+   * @param object the object to serialize and ingest
+   * @param serializer a function that converts the object to a JSON string
+   * @param <T> the type of the object
+   * @throws ZerobusException if the stream is already closed or serialization fails
+   */
+  public <T> void ingestRecordNoWait(T object, JsonSerializer<T> serializer)
+      throws ZerobusException {
+    ensureOpen();
+    String json = serializer.serialize(object);
+    nativeIngestRecordNoWait(nativeHandle, json.getBytes(StandardCharsets.UTF_8), true);
+  }
+
   /**
    * Ingests a JSON string and returns the offset immediately.
    *
@@ -171,6 +193,19 @@ public long ingestRecordOffset(String json) throws ZerobusException {
     return nativeIngestRecordOffset(nativeHandle, json.getBytes(StandardCharsets.UTF_8), true);
   }
 
+  /**
+   * Ingests a JSON string asynchronously without exposing the offset or ack future.
+   *
+   * <p>See {@link #ingestRecordNoWait(Object, JsonSerializer)} for the fire-and-forget contract.
+   *
+   * @param json the JSON string to ingest
+   * @throws ZerobusException if the stream is already closed
+   */
+  public void ingestRecordNoWait(String json) throws ZerobusException {
+    ensureOpen();
+    nativeIngestRecordNoWait(nativeHandle, json.getBytes(StandardCharsets.UTF_8), true);
+  }
+
   // ==================== Batch Ingestion ====================
 
   /**
@@ -220,6 +255,50 @@ public Optional<Long> ingestRecordsOffset(Iterable<String> jsonStrings) throws Z
     return Optional.of(nativeIngestRecordsOffset(nativeHandle, payloads, true));
   }
 
+  /**
+   * Ingests multiple objects as JSON asynchronously without exposing the batch offset.
+   *
+   * <p>See {@link #ingestRecordNoWait(Object, JsonSerializer)} for the fire-and-forget contract.
+   *
+   * @param objects the objects to serialize and ingest
+   * @param serializer a function that converts each object to a JSON string
+   * @param <T> the type of the objects
+   * @throws ZerobusException if the stream is already closed or serialization fails
+   */
+  public <T> void ingestRecordsNoWait(Iterable<T> objects, JsonSerializer<T> serializer)
+      throws ZerobusException {
+    List<byte[]> payloads = new ArrayList<>();
+    for (T obj : objects) {
+      String json = serializer.serialize(obj);
+      payloads.add(json.getBytes(StandardCharsets.UTF_8));
+    }
+    if (payloads.isEmpty()) {
+      return;
+    }
+    ensureOpen();
+    nativeIngestRecordsNoWait(nativeHandle, payloads, true);
+  }
+
+  /**
+   * Ingests multiple JSON strings asynchronously without exposing the batch offset.
+   *
+   * <p>See {@link #ingestRecordNoWait(Object, JsonSerializer)} for the fire-and-forget contract.
+   *
+   * @param jsonStrings the JSON strings to ingest
+   * @throws ZerobusException if the stream is already closed
+   */
+  public void ingestRecordsNoWait(Iterable<String> jsonStrings) throws ZerobusException {
+    List<byte[]> payloads = new ArrayList<>();
+    for (String json : jsonStrings) {
+      payloads.add(json.getBytes(StandardCharsets.UTF_8));
+    }
+    if (payloads.isEmpty()) {
+      return;
+    }
+    ensureOpen();
+    nativeIngestRecordsNoWait(nativeHandle, payloads, true);
+  }
+
   // ==================== Unacknowledged Records ====================
 
   /**

java/src/main/java/com/databricks/zerobus/ZerobusProtoStream.java

@@ -94,6 +94,24 @@ public <T extends Message> long ingestRecordOffset(T record) throws ZerobusExcep
     return nativeIngestRecordOffset(nativeHandle, record.toByteArray(), false);
   }
 
+  /**
+   * Ingests a protobuf message asynchronously without exposing the offset or ack future.
+   *
+   * <p>This fire-and-forget method returns as soon as the record is handed off to the native
+   * runtime. Enqueueing into the stream, native backpressure, and offset assignment happen later on
+   * the runtime's thread pool. Errors raised after this hand-off are intentionally not surfaced to
+   * the caller, and there is no ordering guarantee relative to subsequent offset-returning,
+   * {@link #flush()}, {@link #close()}, or {@link #waitForOffset(long)} calls.
+   *
+   * @param record the protobuf message to ingest
+   * @param <T> the message type
+   * @throws ZerobusException if the stream is already closed or the payload cannot be serialized
+   */
+  public <T extends Message> void ingestRecordNoWait(T record) throws ZerobusException {
+    ensureOpen();
+    nativeIngestRecordNoWait(nativeHandle, record.toByteArray(), false);
+  }
+
   /**
    * Ingests pre-encoded bytes and returns the offset immediately.
    *
@@ -109,6 +127,19 @@ public long ingestRecordOffset(byte[] encodedBytes) throws ZerobusException {
     return nativeIngestRecordOffset(nativeHandle, encodedBytes, false);
   }
 
+  /**
+   * Ingests pre-encoded bytes asynchronously without exposing the offset or ack future.
+   *
+   * <p>See {@link #ingestRecordNoWait(Message)} for the fire-and-forget contract.
+   *
+   * @param encodedBytes the pre-encoded protobuf bytes
+   * @throws ZerobusException if the stream is already closed
+   */
+  public void ingestRecordNoWait(byte[] encodedBytes) throws ZerobusException {
+    ensureOpen();
+    nativeIngestRecordNoWait(nativeHandle, encodedBytes, false);
+  }
+
   // ==================== Batch Ingestion ====================
 
   /**
@@ -152,6 +183,43 @@ public Optional<Long> ingestRecordsOffset(List<byte[]> encodedRecords) throws Ze
     return Optional.of(nativeIngestRecordsOffset(nativeHandle, encodedRecords, false));
   }
 
+  /**
+   * Ingests multiple protobuf messages asynchronously without exposing the batch offset.
+   *
+   * <p>See {@link #ingestRecordNoWait(Message)} for the fire-and-forget contract.
+   *
+   * @param records the protobuf messages to ingest
+   * @param <T> the message type
+   * @throws ZerobusException if the stream is already closed or a payload cannot be serialized
+   */
+  public <T extends Message> void ingestRecordsNoWait(Iterable<T> records) throws ZerobusException {
+    List<byte[]> payloads = new ArrayList<>();
+    for (T record : records) {
+      payloads.add(record.toByteArray());
+    }
+    if (payloads.isEmpty()) {
+      return;
+    }
+    ensureOpen();
+    nativeIngestRecordsNoWait(nativeHandle, payloads, false);
+  }
+
+  /**
+   * Ingests multiple pre-encoded byte arrays asynchronously without exposing the batch offset.
+   *
+   * <p>See {@link #ingestRecordNoWait(Message)} for the fire-and-forget contract.
+   *
+   * @param encodedRecords the pre-encoded protobuf byte arrays
+   * @throws ZerobusException if the stream is already closed
+   */
+  public void ingestRecordsNoWait(List<byte[]> encodedRecords) throws ZerobusException {
+    if (encodedRecords.isEmpty()) {
+      return;
+    }
+    ensureOpen();
+    nativeIngestRecordsNoWait(nativeHandle, encodedRecords, false);
+  }
+
   // ==================== Unacknowledged Records ====================
 
   /**