diff --git a/java/NEXT_CHANGELOG.md b/java/NEXT_CHANGELOG.md index 3867fb00..344dcac7 100644 --- a/java/NEXT_CHANGELOG.md +++ b/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 diff --git a/java/README.md b/java/README.md index c8461906..dd77c119 100644 --- a/java/README.md +++ b/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 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 diff --git a/java/src/main/java/com/databricks/zerobus/StreamBuilder.java b/java/src/main/java/com/databricks/zerobus/StreamBuilder.java index 70f4268b..10bd83d3 100644 --- a/java/src/main/java/com/databricks/zerobus/StreamBuilder.java +++ b/java/src/main/java/com/databricks/zerobus/StreamBuilder.java @@ -180,7 +180,9 @@ public StreamBuilder maxInflightRecords(int maxInflightRecords) { /** * Sets the acknowledgment callback. * - *

Applies to JSON and Protocol Buffer (gRPC) streams. + *

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 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, diff --git a/java/src/test/java/com/databricks/zerobus/StreamBuilderTest.java b/java/src/test/java/com/databricks/zerobus/StreamBuilderTest.java index eafc3405..2a928940 100644 --- a/java/src/test/java/com/databricks/zerobus/StreamBuilderTest.java +++ b/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(); diff --git a/python/NEXT_CHANGELOG.md b/python/NEXT_CHANGELOG.md index 7eeaf5d4..8468ba29 100644 --- a/python/NEXT_CHANGELOG.md +++ b/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 diff --git a/python/tests/test_arrow.py b/python/tests/test_arrow.py index a2e3e2a4..8edd04e0 100644 --- a/python/tests/test_arrow.py +++ b/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): diff --git a/python/zerobus/_zerobus_core.pyi b/python/zerobus/_zerobus_core.pyi index e8042fbc..48d3f504 100644 --- a/python/zerobus/_zerobus_core.pyi +++ b/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: ... diff --git a/rust/NEXT_CHANGELOG.md b/rust/NEXT_CHANGELOG.md index bff28992..c926e11b 100644 --- a/rust/NEXT_CHANGELOG.md +++ b/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 diff --git a/rust/README.md b/rust/README.md index f5f2fc0e..a7d41e34 100644 --- a/rust/README.md +++ b/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` | `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>` | `None` | Optional callback for acknowledgment notifications | -| `callback_max_wait_time_ms` | `Option` | `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>` | `None` | **gRPC JSON/proto streams only.** Optional callback for acknowledgment notifications. Not supported for Arrow Flight streams. | +| `callback_max_wait_time_ms` | `Option` | `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:** diff --git a/rust/sdk/src/builder/stream_builder.rs b/rust/sdk/src/builder/stream_builder.rs index 5e406870..5514667a 100644 --- a/rust/sdk/src/builder/stream_builder.rs +++ b/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) -> 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 { 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() {