[All] Correct SDK documentation and examples

README.md

@@ -139,7 +139,7 @@ Available in the Rust, Python, Go, TypeScript, and Java SDKs starting from their
 - Your workload is naturally columnar or batched — analytics pipelines, gateways aggregating short windows of rows, wide/numeric schemas where row-by-row serialization adds noticeable CPU overhead.
 - Your application already produces Arrow data — pyarrow, the [arrow-rs](https://github.com/apache/arrow-rs) crates, DataFusion, Polars, or other libraries built on Arrow.
 
-For sparse, one-row-at-a-time traffic, JSON or Protocol Buffers over the standard SDK gRPC path are usually simpler. Most SDKs ship a runnable `examples/arrow/` directory; the C++ SDK covers Arrow Flight in its [README](cpp/README.md#arrow-flight-ingestion-beta) until its examples land.
+For sparse, one-row-at-a-time traffic, JSON or Protocol Buffers over the standard SDK gRPC path are usually simpler. Most SDKs ship a runnable `examples/arrow/` directory (see each SDK's README for details).
 
 ### Acknowledgments and throughput
 

cpp/.gitignore

@@ -1,6 +1,7 @@
 # CMake build trees (see Makefile BUILD_DIR; sanitizer builds use build-*).
 /build/
 /build-*/
+/Testing/
 
 # clangd index / editor caches.
 /.cache/

cpp/NEXT_CHANGELOG.md

@@ -18,6 +18,13 @@
 
 ### Documentation
 
+- Corrected custom-header examples and clarified that acknowledgment callbacks
+  run once per logical ingest submission rather than once per record in a batch.
+- Recovery after a flush timeout now treats unacked retrieval failure as an
+  active stream rather than assuming the stream is terminal.
+- Arrow Flight schema validation is documented at stream creation (the schema IPC
+  bytes passed to `create_arrow_stream`), not on the first ingested batch.
+
 ### Internal Changes
 
 ### Breaking Changes

cpp/README.md

@@ -256,7 +256,8 @@ Arrow Flight is **Beta** — the API may change.
 class MyProvider : public zerobus::HeadersProvider {
  public:
   std::map<std::string, std::string> get_headers() override {
-    return {{"Authorization", "Bearer " + current_token()}};
+    return {{"authorization", "Bearer " + current_token()},
+            {"x-databricks-zerobus-table-name", "main.analytics.events"}};
   }
 };
 
@@ -289,7 +290,7 @@ options.ack_callback = zerobus::AckCallback::from(
       // Durable up to `offset` (acks are monotonic: offset N => all <= N acked).
     },
     [](std::int64_t offset, const std::string& msg) noexcept {
-      // The record at `offset` failed terminally.
+      // The logical submission at `offset` failed terminally.
     });
 
 zerobus::Stream stream =
@@ -299,8 +300,8 @@ zerobus::Stream stream =
 Contract (see [`ack_callback.hpp`](include/zerobus/ack_callback.hpp) for the
 canonical version):
 
-- `on_ack` fires once per record in monotonic offset order; `on_error` fires per
-  unacked record on terminal failure (errors also still surface from
+- `on_ack` fires once per logical ingest submission in monotonic offset order;
+  `on_error` fires per unacked submission on terminal failure (errors also still surface from
   `ingest`/`flush`/`wait_for_offset()`).
 - Both methods are **`noexcept`** — an escaping exception crosses the C FFI
   boundary, which is UB, so it calls `std::terminate`. Handle errors inside the

cpp/examples/arrow/README.md

@@ -94,7 +94,8 @@ stream.close();
 - **All-or-nothing per `RecordBatch`** — a batch is acknowledged as a unit.
 - **Single acknowledgment** — one offset ID for the whole `RecordBatch`.
 - **Schema validation** — the `RecordBatch` schema must match the schema
-  configured on the stream. The server validates on the first batch and fails
+  configured on the stream. The server validates that schema when the stream is
+  created, from the schema IPC bytes passed to `create_arrow_stream`, and fails
   fast with a descriptive error on a mismatch.
 
 ## IPC Compression

cpp/examples/arrow/arrow_ingest.cpp

@@ -73,8 +73,9 @@ std::int64_t now_micros() {
 // type. This mirrors the canonical Arrow schema the Databricks Arrow Flight
 // server derives from a Delta table: Delta STRING -> large_utf8, INT -> int32,
 // DOUBLE -> float64, TIMESTAMP -> timestamp(microsecond, "UTC"). The server
-// validates the record-batch schema on the first batch and fails fast with a
-// descriptive error on a mismatch.
+// validates the schema when the stream is created, from the schema IPC bytes
+// passed to create_arrow_stream, and fails fast with a descriptive error on a
+// mismatch.
 std::shared_ptr<arrow::Schema> orders_schema() {
   auto utc_micros = arrow::timestamp(arrow::TimeUnit::MICRO, "UTC");
   return arrow::schema({

cpp/examples/json/README.md

@@ -92,15 +92,23 @@ zerobus::Stream stream =
 transient disconnects. If a stream fails *terminally*, `flush()`/`close()`
 throws — and a failed `close()` keeps the handle alive so you can drain whatever
 was never acknowledged with `get_unacked_records()` and re-ingest it on a fresh
-stream. (After a *successful* `close()` the handle is freed, so that call would
-throw instead — recovery belongs on the failure path only.)
+stream. A flush timeout can leave the stream active; retrieval then throws, and
+those records cannot be recovered until the stream has actually closed. After a
+*successful* `close()` the handle is freed, so that call would throw instead —
+recovery belongs on the failure path only.
 
 ```cpp
 try {
   stream.flush();
   stream.close();
 } catch (const zerobus::ZerobusException& e) {
-  std::vector<zerobus::UnackedRecord> unacked = stream.get_unacked_records();
+  std::vector<zerobus::UnackedRecord> unacked;
+  try {
+    unacked = stream.get_unacked_records();
+  } catch (const zerobus::ZerobusException& retrieval) {
+    // Stream may still be active (for example a flush timeout).
+    throw;
+  }
   zerobus::Stream retry = open_stream(...);
   for (const auto& record : unacked) {
     retry.ingest_json_record(record.as_string());   // loop — no per-record wait
@@ -130,7 +138,7 @@ Each `UnackedRecord` exposes `is_json()`, the raw `data()` bytes, and
 ```
 Batch of 3 records queued; batch offset ID: 0
 Batch acknowledged at offset ID: 0
-Stream closed successfully. Callback observed 3 acknowledgements.
+Stream closed successfully. Callback observed 1 logical submission acknowledgement(s).
 ```
 
 ### Code Highlights

cpp/examples/json/batch.cpp

@@ -74,10 +74,10 @@ std::string make_order_json(int id, const std::string& customer,
 // core calls get_headers() whenever it needs fresh headers (possibly from
 // another thread), and you return whatever the endpoint expects — at minimum an
 // "authorization" bearer token and "x-databricks-zerobus-table-name". Throwing
-// surfaces the message to the core as a headers-provider error. The provider
-// must outlive the Stream (which holds a shared_ptr to it). See
-// include/zerobus/headers_provider.hpp for the full contract. Used in the
-// commented-out create_stream() call below.
+// surfaces the message to the core as a headers-provider error. Provider
+// ownership is handed to the FFI, so the caller does not need to retain its own
+// shared_ptr after stream creation. See include/zerobus/headers_provider.hpp
+// for the full contract. Used in the commented-out create_stream() call below.
 class BearerTokenProvider : public zerobus::HeadersProvider {
  public:
   BearerTokenProvider(std::string table_name, std::string token)
@@ -136,7 +136,7 @@ int main() {
           (void)offset;
         },
         [](std::int64_t offset, const std::string& msg) noexcept {
-          std::cerr << "record at offset " << offset << " failed: " << msg
+          std::cerr << "batch at offset " << offset << " failed: " << msg
                     << "\n";
         });
     options.callback_wait_policy = zerobus::CallbackWaitPolicy::forever();
@@ -183,7 +183,7 @@ int main() {
     stream.flush();
     stream.close();
     std::cout << "Stream closed successfully. Callback observed "
-              << acked.load() << " acknowledgements.\n";
+              << acked.load() << " logical submission acknowledgement(s).\n";
   } catch (const zerobus::ZerobusException& e) {
     std::cerr << "Zerobus error: " << e.what()
               << " (retryable=" << (e.is_retryable() ? "true" : "false")

cpp/examples/json/single.cpp

@@ -141,8 +141,15 @@ int main() {
     } catch (const zerobus::ZerobusException& e) {
       std::cerr << "Stream failed: " << e.what() << "\n";
 
-      std::vector<zerobus::UnackedRecord> unacked =
-          stream.get_unacked_records();
+      std::vector<zerobus::UnackedRecord> unacked;
+      try {
+        unacked = stream.get_unacked_records();
+      } catch (const zerobus::ZerobusException& retrieval) {
+        std::cerr << "Could not inspect unacked records (stream may still be "
+                     "active): "
+                  << retrieval.what() << "\n";
+        return 1;
+      }
       std::cout << "Recovering " << unacked.size()
                 << " unacknowledged records on a fresh stream.\n";
 

cpp/include/zerobus/ack_callback.hpp

@@ -12,9 +12,10 @@ namespace zerobus {
 /// in `wait_for_offset()` / `flush()`. Register via
 /// `StreamOptions::ack_callback`.
 ///
-/// `on_ack` fires once per record, in monotonic offset order (offset `N` =>
-/// all `<= N` acked); `on_error` fires per unacked record on terminal failure,
-/// which may also surface from `ingest`/`flush`/`wait_for_offset()`. Callbacks
+/// `on_ack` fires once per logical ingest submission, in monotonic offset order
+/// (offset `N` => all `<= N` acked); `on_error` fires per unacked submission on
+/// terminal failure, which may also surface from `ingest`/`flush`/
+/// `wait_for_offset()`. Callbacks
 /// run serialized on a background task, possibly on another thread: synchronize
 /// shared state, keep them light, and don't call back into the owning `Stream`
 /// (that is concurrent use of a non-thread-safe object).
@@ -38,12 +39,13 @@ class AckCallback {
  public:
   virtual ~AckCallback() = default;
 
-  /// Called when the record at @p offset has been durably acknowledged.
+  /// Called when the logical submission at @p offset has been durably
+  /// acknowledged.
   virtual void on_ack(std::int64_t offset) noexcept = 0;
 
-  /// Called when the record at @p offset failed terminally.
+  /// Called when the logical submission at @p offset failed terminally.
   ///
-  /// @param offset The logical offset of the failed record.
+  /// @param offset The logical offset of the failed submission.
   /// @param error_message Human-readable error text from the core.
   virtual void on_error(std::int64_t offset,
                         const std::string& error_message) noexcept = 0;

cpp/include/zerobus/stream.hpp

@@ -105,7 +105,8 @@ class Stream {
   /// Return all unacknowledged records from a closed or failed stream, for the
   /// caller to re-ingest on a fresh stream. Remains callable after a failed
   /// `close()` (which keeps the handle alive precisely so recovery is
-  /// possible).
+  /// possible). Calling on an active stream (e.g. after a flush timeout before
+  /// the stream has closed) throws ZerobusException.
   ///
   /// @return The records that were ingested but not acknowledged.
   /// @throws ZerobusException if the records cannot be retrieved.

dotnet/CONTRIBUTING.md

@@ -9,7 +9,7 @@ This document covers .NET-specific development setup and workflow.
 ### Prerequisites
 
 - Git
-- .NET SDK 8.0 or higher
+- .NET SDK 10.0 or higher (the projects target both .NET 8 and .NET 10)
 - Rust toolchain (`cargo`) - [Install Rust](https://rustup.rs/)
 - Bash shell (used by `build_native.sh`)
 
@@ -98,7 +98,7 @@ When making FFI-related changes:
 
 1. Update Rust code in `../rust/ffi/src/`
 2. Update exported C API in `../rust/ffi/zerobus.h` if needed
-3. Update .NET interop bindings in `src/Zerobus/Interop/`
+3. Update .NET interop bindings in `src/Zerobus/Native/`
 4. Rebuild native artifacts:
 	```bash
 	./build_native.sh

dotnet/NEXT_CHANGELOG.md

@@ -21,6 +21,13 @@
 
 ### Documentation
 
+- Corrected installation and source-build prerequisites, separated JSON and
+  Protobuf stream examples, added a runnable generated-message example, and
+  replaced per-record waits with one final flush in bulk-ingestion examples.
+- Documented that `GetUnackedRecords()` can fail while the stream is still
+  active after a flush timeout, and stopped reporting success after ingest
+  failures. Pointed CONTRIBUTING at `src/Zerobus/Native/`.
+
 ### Internal Changes
 
 - Made the .NET release workflow build-only, consistent with the other SDKs. It now packs the NuGet package and uploads it as an artifact; publishing and the GitHub Release happen downstream.

dotnet/README.md

@@ -4,8 +4,8 @@ High-performance .NET SDK for streaming data ingestion into Databricks Delta tab
 
 ## Requirements
 
-- **.NET 8** or **.NET 10**
-- **Rust toolchain** (for building the native `zerobus_ffi` library from source)
+- Consumers: .NET 8 or .NET 10
+- Building from source: .NET 10 SDK and a Rust toolchain
 
 ## Quick Start
 
@@ -31,11 +31,12 @@ using var stream = sdk.CreateJsonStream(
     clientSecret,
     options);
 
-// 4. Ingest records.
-long offset = stream.IngestRecord("""{"id": 1, "message": "Hello"}""");
-
-// 5. Wait for acknowledgment.
-stream.WaitForOffset(offset);
+// 4. Queue records, then confirm the whole run with one flush.
+for (int id = 1; id <= 100; id++)
+{
+    stream.IngestRecord($$"""{"id": {{id}}, "message": "Hello"}""");
+}
+stream.Flush();
 ```
 
 ## Installation
@@ -187,13 +188,26 @@ If you use this untyped API, JSON streams must set `RecordType.Json` and use the
 Proto streams must provide `DescriptorProto` and use the byte-oriented overloads.
 
 ```csharp
-// JSON
-long offset = stream.IngestRecord("""{"field": "value"}""");
+// JSON stream
+var jsonOptions = options with { RecordType = RecordType.Json };
+using var jsonStream = sdk.CreateStream(
+    new TableProperties("catalog.schema.json_table"),
+    clientId,
+    clientSecret,
+    jsonOptions);
+long jsonOffset = jsonStream.IngestRecord("""{"field": "value"}""");
+jsonStream.WaitForOffset(jsonOffset);
 
-// Protobuf
+// Protobuf stream
+var protoOptions = options with { RecordType = RecordType.Proto };
 byte[] protoBytes = myMessage.ToByteArray();
-long offset = stream.IngestRecord(protoBytes);
-stream.WaitForOffset(offset);
+using var protoStream = sdk.CreateStream(
+    new TableProperties("catalog.schema.proto_table", descriptorProto),
+    clientId,
+    clientSecret,
+    protoOptions);
+long protoOffset = protoStream.IngestRecord(protoBytes);
+protoStream.WaitForOffset(protoOffset);
 ```
 
 #### `IngestRecords`
@@ -205,8 +219,8 @@ string[] records = [
     """{"device": "sensor-001", "temp": 20}""",
     """{"device": "sensor-002", "temp": 21}""",
 ];
-long lastOffset = stream.IngestRecords(records);
-stream.WaitForOffset(lastOffset);
+long batchOffset = stream.IngestRecords(records);
+stream.WaitForOffset(batchOffset);
 ```
 
 #### `WaitForOffset` (sync)
@@ -227,14 +241,26 @@ stream.Flush();
 
 #### `GetUnackedRecords`
 
-Retrieves unacknowledged records after stream failure (call after close/failure only).
+Retrieves unacknowledged records after stream failure (call after close/failure only). A flush timeout can leave the stream active, in which case `GetUnackedRecords()` throws until the stream closes.
 
 ```csharp
-ReadOnlyMemory<byte>[] unacked = stream.GetUnackedRecords();
-foreach (var payload in unacked)
+try
+{
+    stream.Flush();
+}
+catch (ZerobusException)
 {
-    // Decode as UTF-8 if you know this stream ingests JSON.
-    Console.WriteLine($"record bytes: {payload.Length}");
+    // A flush timeout can leave the stream active. GetUnackedRecords
+    // requires a closed or failed stream.
+    try
+    {
+        var unacked = stream.GetUnackedRecords();
+        Console.WriteLine($"Failed to acknowledge {unacked.Length} records");
+    }
+    catch (ZerobusException retrieval)
+    {
+        Console.WriteLine($"Could not inspect unacked records (stream may still be active): {retrieval.Message}");
+    }
 }
 ```
 

dotnet/examples/JsonBatch/Program.cs

@@ -42,11 +42,11 @@
     """{"device_name": "sensor-005", "temp": 24, "humidity": 64}""",
 ];
 
-long lastOffset = stream.IngestRecords(batchRecords);
-Console.WriteLine($"Batch of {batchRecords.Length} records ingested, last offset: {lastOffset}");
+long batchOffset = stream.IngestRecords(batchRecords);
+Console.WriteLine($"Batch of {batchRecords.Length} records ingested, batch offset: {batchOffset}");
 
-// Wait for the last offset to ensure the entire batch is acknowledged.
-stream.WaitForOffset(lastOffset);
+// Wait for the batch offset to ensure the entire batch is acknowledged.
+stream.WaitForOffset(batchOffset);
 Console.WriteLine("Batch acknowledged!");
 
 Console.WriteLine("All operations completed successfully!");