[ArrowFlight] Stream builder - Reject unsupported Arrow ACK callbacks

java/NEXT_CHANGELOG.md

@@ -10,6 +10,10 @@
 
 ### Bug Fixes
 
+- Arrow builders now reject unsupported ACK callbacks instead of silently
+  discarding them. Configuring `ackCallback` before calling `ArrowStreamBuilder.build()`
+  throws `IllegalStateException`.
+
 ### Documentation
 
 ### Internal Changes

java/README.md

@@ -910,6 +910,8 @@ Used with `ZerobusArrowStream`. Build via `ArrowStreamConfigurationOptions.build
 | `ipcCompression` | `IPCCompressionType.NONE` | Arrow IPC compression codec on the wire (`NONE`, `LZ4_FRAME`, `ZSTD`) |
 | `streamPausedMaxWaitTimeMs` | `-1` | Max time to wait in the paused state during graceful close. `-1` = full server duration, `0` = immediate recovery, `>0` = `min(this, server_duration)` ms |
 
+`ackCallback` is not supported for Arrow Flight streams. Configuring it on `StreamBuilder` before calling `ArrowStreamBuilder.build()` throws `IllegalStateException`.
+
 ## Logging
 
 The Databricks Zerobus Ingest SDK for Java uses the standard [SLF4J logging framework](https://www.slf4j.org/). The SDK only depends on `slf4j-api`, which means **you need to add an SLF4J implementation** to your classpath to see log output.
@@ -1171,6 +1173,8 @@ CompletableFuture<ZerobusArrowStream> StreamBuilder.ArrowStreamBuilder.build()
 
 `ArrowStreamBuilder` additionally supports `maxInflightBatches(int)`, `connectionTimeoutMs(long)`, `ipcCompression(IPCCompressionType)`, and `streamPausedMaxWaitTimeMs(long)`.
 
+Configuring `ackCallback` before calling `ArrowStreamBuilder.build()` throws `IllegalStateException` because Arrow Flight streams do not support ACK callbacks.
+
 ---
 
 ### ZerobusProtoStream

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

@@ -180,7 +180,9 @@ public StreamBuilder maxInflightRecords(int maxInflightRecords) {
   /**
    * Sets the acknowledgment callback.
    *
-   * <p>Applies to JSON and Protocol Buffer (gRPC) streams.
+   * <p>Applies to JSON and Protocol Buffer (gRPC) streams only. Arrow Flight streams do not support
+   * ACK callbacks; configuring one and calling {@link ArrowStreamBuilder#build()} throws {@link
+   * IllegalStateException}.
    *
    * @param ackCallback the acknowledgment callback
    * @return this builder for method chaining
@@ -469,10 +471,14 @@ ArrowStreamConfigurationOptions buildOptions() {
      * Builds and opens the Arrow Flight stream.
      *
      * @return a future that completes with the {@link ZerobusArrowStream} when ready
-     * @throws IllegalStateException if the table name or authentication has not been set
+     * @throws IllegalStateException if the table name or authentication has not been set, or if an
+     *     ACK callback was configured via {@link StreamBuilder#ackCallback(AckCallback)}
      */
     public CompletableFuture<ZerobusArrowStream> build() {
       base.validateRequired();
+      if (base.ackCallback != null) {
+        throw new IllegalStateException("ackCallback is not supported for Arrow Flight streams");
+      }
       return base.sdk.createArrowStreamInternal(
           base.tableName,
           schema,

java/src/test/java/com/databricks/zerobus/StreamBuilderTest.java

@@ -327,6 +327,28 @@ void compiledProtoBuildRoutesToCreateProtoStreamInternal() {
     verify(sdk, never()).createArrowStreamInternal(any(), any(), any(), any(), any(), any());
   }
 
+  @Test
+  void arrowBuildRejectsAckCallback() {
+    AckCallback callback =
+        new AckCallback() {
+          @Override
+          public void onAck(long offsetId) {}
+
+          @Override
+          public void onError(long offsetId, String errorMessage) {}
+        };
+
+    StreamBuilder.ArrowStreamBuilder arrowBuilder =
+        builder()
+            .table("catalog.schema.table")
+            .oauth("client-id", "client-secret")
+            .ackCallback(callback)
+            .arrow(emptySchema());
+
+    IllegalStateException ex = assertThrows(IllegalStateException.class, arrowBuilder::build);
+    assertEquals("ackCallback is not supported for Arrow Flight streams", ex.getMessage());
+  }
+
   @Test
   void arrowBuildRoutesToCreateArrowStreamInternal() {
     assumeNativeLibrary();

python/NEXT_CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Documentation
 
+- Document `stream_paused_max_wait_time_ms` on `ArrowStreamConfigurationOptions` in the Arrow type stub so it matches the already-supported runtime option.
+
 ### Internal Changes
 
 ### Breaking Changes

python/tests/test_arrow.py

@@ -180,6 +180,7 @@ def test_default_construction(self):
         self.assertIsInstance(options.flush_timeout_ms, int)
         self.assertIsInstance(options.connection_timeout_ms, int)
         self.assertEqual(options.ipc_compression, IPCCompression.NONE)
+        self.assertIsNone(options.stream_paused_max_wait_time_ms)
 
     def test_kwargs_construction(self):
         options = ArrowStreamConfigurationOptions(
@@ -201,6 +202,7 @@ def test_all_kwargs(self):
             server_lack_of_ack_timeout_ms=30000,
             flush_timeout_ms=5000,
             connection_timeout_ms=8000,
+            stream_paused_max_wait_time_ms=2500,
         )
         self.assertEqual(options.max_inflight_batches, 20)
         self.assertTrue(options.recovery)
@@ -210,6 +212,7 @@ def test_all_kwargs(self):
         self.assertEqual(options.server_lack_of_ack_timeout_ms, 30000)
         self.assertEqual(options.flush_timeout_ms, 5000)
         self.assertEqual(options.connection_timeout_ms, 8000)
+        self.assertEqual(options.stream_paused_max_wait_time_ms, 2500)
 
     def test_unknown_kwarg_raises(self):
         with self.assertRaises(ValueError):
@@ -219,8 +222,10 @@ def test_setters(self):
         options = ArrowStreamConfigurationOptions()
         options.max_inflight_batches = 99
         options.recovery = False
+        options.stream_paused_max_wait_time_ms = 1500
         self.assertEqual(options.max_inflight_batches, 99)
         self.assertFalse(options.recovery)
+        self.assertEqual(options.stream_paused_max_wait_time_ms, 1500)
 
     def test_ipc_compression_lz4(self):
         options = ArrowStreamConfigurationOptions(ipc_compression=IPCCompression.LZ4_FRAME)
@@ -257,6 +262,7 @@ def test_repr(self):
         self.assertIn("max_inflight_batches", repr_str)
         self.assertIn("recovery", repr_str)
         self.assertIn("ipc_compression", repr_str)
+        self.assertIn("stream_paused_max_wait_time_ms", repr_str)
 
 
 class TestSerializeBatchEmptyRecordBatch(unittest.TestCase):

python/zerobus/_zerobus_core.pyi

@@ -582,6 +582,13 @@ class arrow:
         ipc_compression: "arrow.IPCCompression"
         """IPC compression codec. Default: IPCCompression.None"""
 
+        stream_paused_max_wait_time_ms: Optional[int]
+        """Maximum ACK-wait cap during server-initiated graceful rotation.
+        None uses the server grace period available after reserving bounded
+        transport-cleanup time, 0 skips ACK waiting, and positive values cap the
+        ACK wait at min(value, remaining grace) milliseconds. Bounded transport
+        cleanup still runs when ACK waiting is skipped. Default: None"""
+
         def __init__(
             self,
             *,
@@ -594,6 +601,7 @@ class arrow:
             flush_timeout_ms: int = 300000,
             connection_timeout_ms: int = 30000,
             ipc_compression: "arrow.IPCCompression" = ...,
+            stream_paused_max_wait_time_ms: Optional[int] = None,
         ) -> None: ...
         def __repr__(self) -> str: ...
 

rust/NEXT_CHANGELOG.md

@@ -8,6 +8,9 @@
 
 ### Bug Fixes
 
+- Arrow builders now reject unsupported ACK callbacks instead of silently
+  discarding them. Remove `.ack_callback(...)` before calling `build_arrow()`;
+  otherwise it returns `InvalidArgument`.
 - Arrow Flight acknowledgment deadlines are pending-relative: no timer runs
   while a stream is idle. During normal stream operation, each batch receives
   an absolute deadline when it becomes pending; responses and partial

rust/README.md

@@ -802,8 +802,8 @@ Also accepts `0` or `no`.
 | `flush_timeout_ms` | `u64` | 300,000 | Timeout for flush operations (ms) |
 | `record_type` | `RecordType` | `RecordType::Proto` | Record serialization format (Proto or Json) |
 | `stream_paused_max_wait_time_ms` | `Option<u64>` | `None` | Max time to wait for outstanding acknowledgments during graceful close (`None` = server grace remaining after reserving transport cleanup time, `Some(0)` = skip the ACK wait, `Some(x)` = the smaller of `x` and that remaining grace). A bounded request/response drain still runs after the ACK wait. |
-| `ack_callback` | `Option<Arc<dyn AckCallback>>` | `None` | Optional callback for acknowledgment notifications |
-| `callback_max_wait_time_ms` | `Option<u64>` | `None` | Maximum time to wait for callback processing to complete after closing the stream (`None` = wait indefinitely, `Some(x)` = wait up to `x` ms) |
+| `ack_callback` | `Option<Arc<dyn AckCallback>>` | `None` | **gRPC JSON/proto streams only.** Optional callback for acknowledgment notifications. Not supported for Arrow Flight streams. |
+| `callback_max_wait_time_ms` | `Option<u64>` | `Some(5_000)` | **gRPC JSON/proto streams only.** Maximum time to wait for callback processing to complete after closing the stream (`None` = wait indefinitely, `Some(x)` = wait up to `x` ms) |
 
 For Arrow Flight streams *(Beta)*, `server_lack_of_ack_timeout_ms` is an
 absolute limit on how long the oldest batch may remain pending during normal
@@ -818,7 +818,9 @@ within the timeout.
 Arrow stream construction rejects `recovery_timeout_ms` and
 `server_lack_of_ack_timeout_ms` values whose deadlines cannot be represented by
 the platform monotonic clock. Server-advertised graceful-rotation periods are
-capped at one year.
+capped at one year. If `ack_callback` was configured on the builder,
+`build_arrow()` returns `ZerobusError::InvalidArgument` instead of silently
+ignoring the callback.
 
 **Example:**
 

rust/sdk/src/builder/stream_builder.rs

@@ -314,6 +314,10 @@ impl<'a> StreamBuilder<'a> {
     }
 
     /// Set the acknowledgment callback (gRPC streams only).
+    ///
+    /// `build_arrow()` rejects this option with
+    /// [`ZerobusError::InvalidArgument`] because Arrow Flight streams do not
+    /// support acknowledgment callbacks.
     pub fn ack_callback(mut self, callback: Arc<dyn AckCallback>) -> Self {
         self.grpc_config.ack_callback = Some(callback);
         self
@@ -457,11 +461,33 @@ impl<'a> StreamBuilder<'a> {
     /// Build and open an Arrow Flight ingestion stream.
     ///
     /// Returns an error if table name, authentication, or format has not been set,
-    /// or if a non-Arrow format was selected (use `build()` instead).
+    /// if a non-Arrow format was selected (use `build()` instead), or if
+    /// [`ack_callback`](Self::ack_callback) was configured (callbacks are
+    /// unsupported for Arrow Flight streams).
     #[cfg(feature = "arrow-flight")]
     pub async fn build_arrow(self) -> ZerobusResult<ZerobusArrowStream> {
         self.validate()?;
 
+        let schema = match self.format.as_ref() {
+            Some(FormatConfig::Arrow(schema)) => Arc::clone(schema),
+            Some(_) => {
+                return Err(ZerobusError::InvalidArgument(
+                    "non-Arrow format requires .build() instead of .build_arrow()".into(),
+                ));
+            }
+            None => {
+                return Err(ZerobusError::InvalidArgument(
+                    "record format is required: call .arrow() before .build_arrow()".into(),
+                ));
+            }
+        };
+
+        if self.grpc_config.ack_callback.is_some() {
+            return Err(ZerobusError::InvalidArgument(
+                "ack_callback is not supported for Arrow Flight streams".into(),
+            ));
+        }
+
         // Arrow-only: a zero bound deadlocks ingest / panics the channel. Not in the
         // shared validate() since it's irrelevant to JSON/proto build().
         if self.arrow_config.max_inflight_batches == 0 {
@@ -480,20 +506,6 @@ impl<'a> StreamBuilder<'a> {
 
         let headers_provider = self.resolve_headers_provider()?;
 
-        let schema = match self.format {
-            Some(FormatConfig::Arrow(schema)) => schema,
-            Some(_) => {
-                return Err(ZerobusError::InvalidArgument(
-                    "non-Arrow format requires .build() instead of .build_arrow()".into(),
-                ));
-            }
-            None => {
-                return Err(ZerobusError::InvalidArgument(
-                    "record format is required: call .arrow() before .build_arrow()".into(),
-                ));
-            }
-        };
-
         let table_properties = ArrowTableProperties {
             table_name: self.table_name,
             schema,
@@ -520,6 +532,16 @@ mod tests {
     use super::*;
     use std::collections::HashMap;
 
+    #[cfg(feature = "arrow-flight")]
+    struct NoopAckCallback;
+
+    #[cfg(feature = "arrow-flight")]
+    impl AckCallback for NoopAckCallback {
+        fn on_ack(&self, _offset_id: crate::offset_generator::OffsetId) {}
+
+        fn on_error(&self, _offset_id: crate::offset_generator::OffsetId, _error_message: &str) {}
+    }
+
     fn test_sdk() -> ZerobusSdk {
         ZerobusSdk::new_with_config(
             "http://localhost:1234".to_string(),
@@ -791,6 +813,59 @@ mod tests {
             .connection_timeout_ms(10_000);
     }
 
+    #[cfg(feature = "arrow-flight")]
+    #[tokio::test]
+    async fn arrow_builder_rejects_ack_callback() {
+        use arrow_schema::{DataType, Field, Schema as ArrowSchema};
+
+        let sdk = test_sdk();
+        let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+            "id",
+            DataType::Int32,
+            false,
+        )]));
+        let result = sdk
+            .stream_builder()
+            .table("t")
+            .oauth("a", "b")
+            .arrow(schema)
+            .ack_callback(Arc::new(NoopAckCallback))
+            .build_arrow()
+            .await;
+
+        match result {
+            Err(ZerobusError::InvalidArgument(msg)) => {
+                assert!(msg.contains("ack_callback"));
+                assert!(msg.contains("Arrow Flight"));
+            }
+            _ => panic!("expected InvalidArgument error"),
+        }
+    }
+
+    #[cfg(feature = "arrow-flight")]
+    #[tokio::test]
+    async fn arrow_builder_reports_format_error_before_ack_callback_error() {
+        let sdk = test_sdk();
+        let result = sdk
+            .stream_builder()
+            .table("t")
+            .oauth("a", "b")
+            .json()
+            .ack_callback(Arc::new(NoopAckCallback))
+            .build_arrow()
+            .await;
+
+        match result {
+            Err(ZerobusError::InvalidArgument(msg)) => {
+                assert_eq!(
+                    msg,
+                    "non-Arrow format requires .build() instead of .build_arrow()"
+                );
+            }
+            _ => panic!("expected non-Arrow format InvalidArgument error"),
+        }
+    }
+
     #[cfg(feature = "arrow-flight")]
     #[test]
     fn shared_setters_write_to_arrow_config() {