feldera · swanandx · Jul 13, 2026 · Jul 13, 2026 · Jul 9, 2026 · Jul 9, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/adapterlib/Cargo.toml b/crates/adapterlib/Cargo.toml
@@ -39,14 +39,17 @@ rmpv = { workspace = true, features = ["with-serde"] }
 bytemuck = { workspace = true }
 num-traits = { workspace = true }
 num-derive = { workspace = true }
-tokio = { workspace = true, features = ["sync"] }
+tokio = { workspace = true, features = ["sync", "rt"] }
 xxhash-rust = { workspace = true, features = ["xxh3"] }
 datafusion = { workspace = true }
 datafusion-functions-json = { workspace = true }
 thiserror = { workspace = true }
 serde_json_path_to_error = { workspace = true }
 chrono = { workspace = true }
 tracing = { workspace = true }
+async-channel = { workspace = true }
+futures = { workspace = true }
+rand = { workspace = true }
 
 [package.metadata.cargo-machete]
 ignored = ["num-traits"]
diff --git a/crates/adapterlib/src/utils.rs b/crates/adapterlib/src/utils.rs
@@ -1 +1,3 @@
+pub mod backoff;
 pub mod datafusion;
+pub mod job_queue;
diff --git a/crates/adapterlib/src/utils/backoff.rs b/crates/adapterlib/src/utils/backoff.rs
@@ -0,0 +1,46 @@
+//! Exponential backoff with jitter, shared by connectors that retry
+//! intermittent object-store and metadata failures.
+
+use rand::Rng;
+use std::cmp::min;
+use std::time::Duration;
+
+/// Calculate exponential backoff delay for retrying an operation.
+///
+/// Starts at 0.5s, doubles each retry, caps at 32s, plus uniform jitter up to 25% of that delay
+/// (capped at the 32s ceiling) to reduce synchronized retries across connectors.
+pub fn calculate_backoff_delay(retry_count: u32) -> Duration {
+    let base_delay_ms: u64 = 500; // 0.5 seconds
+    let max_delay_ms: u64 = 32_000; // 32 seconds
+    let delay_ms = min(
+        base_delay_ms.checked_shl(retry_count).unwrap_or(u64::MAX),
+        max_delay_ms,
+    );
+    let jitter_span = (delay_ms / 4).max(1);
+    let jitter_ms = rand::thread_rng().gen_range(0..jitter_span);
+    Duration::from_millis(min(delay_ms + jitter_ms, max_delay_ms))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::calculate_backoff_delay;
+    use std::time::Duration;
+
+    #[test]
+    fn backoff_grows_and_caps() {
+        // Delay lower bound doubles with each retry until the 32s cap.
+        // Upper bound adds up to 25% jitter, itself capped at 32s.
+        for (retry, min_ms) in [(0u32, 500u64), (1, 1000), (2, 2000), (6, 32000)] {
+            let d = calculate_backoff_delay(retry);
+            assert!(d >= Duration::from_millis(min_ms), "retry {retry}: {d:?}");
+            assert!(d <= Duration::from_millis(32_000), "retry {retry}: {d:?}");
+        }
+    }
+
+    #[test]
+    fn backoff_saturates_on_large_retry_count() {
+        // `checked_shl` overflow must not panic; it saturates to the cap.
+        let d = calculate_backoff_delay(1000);
+        assert!(d <= Duration::from_millis(32_000));
+    }
+}
diff --git a/crates/adapterlib/src/utils/job_queue.rs b/crates/adapterlib/src/utils/job_queue.rs
@@ -0,0 +1,205 @@
+//! A job queue that dispatches work to a pool of tokio tasks while preserving
+//! output ordering.
+//!
+//! Connectors use this to parse record batches in parallel: several worker
+//! tasks parse concurrently, but their results are consumed in the order the
+//! jobs were enqueued, so records reach the circuit in their original order.
+
+use std::future::Future;
+use std::pin::Pin;
+
+use futures::channel::oneshot;
+use tokio::{spawn, task::JoinHandle};
+
+/// A job queue that dispatches work to a pool of tokio tasks.
+///
+/// While the jobs can complete out-of-order, their outputs are consumed in the same order
+/// they were enqueued. This is useful for implementing parallel parsing, where parsed
+/// records must be fed into the circuit in the order they were received.
+/*
+                          sync
+   ┌─────────────────────────────────────────────────────────┐
+   │                           ┌───────┐                     │
+   │                        ┌─►│worker1├────────┐            │
+   │              jobs      │  └───────┘        │            │
+   ▼          ┌─┬─┬─┬─┬─┬─┐ │  ┌───────┐        │       ┌────┴───┐
+producer  ─┬─►│ │ │ │ │ │ ├─┼─►│worker2├────┐   │       │consumer│
+           │  └─┴─┴─┴─┴─┴─┘ │  └───────┘    │   │       └────────┘
+           │                │  ┌───────┐    │   │            ▲
+           │                └─►│worker3├──┐ │   │            │
+           │                   └───────┘  ▼ ▼   ▼            │
+           │                             ┌─┬─┬─┬─┬─┬─┐       │
+           └────────────────────────────►│ │ │ │ │ │ ├───────┘
+                                         └─┴─┴─┴─┴─┴─┘
+                                          completions
+
+* Producer adds jobs to the job queue. A job consists of an input value and
+  a completion one-shot channel where the worker will send the output of the job.
+
+* The receiving side of the channel is pushed to the completions queue.
+
+* Worker tasks dequeue jobs from the jobs queue and send the result to the
+  one-shot channel associated with each job. The consumer receives the next
+  item from the completion queue and waits for the output of the job.
+
+* When the producer needs to wait for all jobs in the queue to complete, it
+  sends a special Sync message to the completions queue. Upon receiving
+  this message, the consumer sends an acknowledgement to the sync channel.
+*/
+pub struct JobQueue<I, O> {
+    /// The producer side of the job queue.
+    job_sender: async_channel::Sender<(I, oneshot::Sender<O>)>,
+
+    /// The producer side of the completions queue.
+    completion_sender: async_channel::Sender<Completion<O>>,
+
+    /// The receiving side of the sync channel.
+    sync_receiver: async_channel::Receiver<()>,
+
+    /// Worker tasks.
+    workers: Vec<JoinHandle<()>>,
+
+    /// Consumer task.
+    consumer: JoinHandle<()>,
+}
+
+// Every message in the completions channel contains the receiving side of the
+// oneshot channel that will contain the result of the completed job, or a special
+// Sync message.
+enum Completion<O> {
+    Completion(oneshot::Receiver<O>),
+    Sync,
+}
+
+impl<I, O> JobQueue<I, O>
+where
+    I: Send + 'static,
+    O: Send + 'static,
+{
+    /// Create a job queue.
+    ///
+    /// # Arguments
+    ///
+    /// * `num_workers` - the number of threads in the worker pool. Must be >0.
+    /// * `worker_builder` - a closure that returns a closure that each worker will execute for each job.
+    ///   The outer closure is needed to allocate any resources needed by the worker.
+    /// * `consumer_func` - closure that the consumer will execute for each completed job.
+    pub fn new(
+        num_workers: usize,
+        worker_builder: impl Fn() -> Box<dyn FnMut(I) -> Pin<Box<dyn Future<Output = O> + Send>> + Send>,
+        mut consumer_func: impl FnMut(O) + Send + 'static,
+    ) -> Self {
+        assert_ne!(num_workers, 0);
+
+        // The jobs queue length is equal to the number of workers. This way, workers don't get
+        // starved, but we also don't queue more data than necessary to keep the workers busy.
+        // TODO: does it make sense to make queue length separately configurable?
+        let (job_sender, job_receiver) =
+            async_channel::bounded::<(I, oneshot::Sender<O>)>(num_workers);
+
+        // The completion queue can contain at most one message per worker + one message per
+        // job in the jobs queue plus the Sync message.
+        let (completion_sender, completion_receiver) = async_channel::bounded(2 * num_workers + 1);
+        let (sync_sender, sync_receiver) = async_channel::bounded(1);
+
+        let workers = (0..num_workers)
+            .map(move |_| {
+                let mut worker_fn = worker_builder();
+
+                let job_receiver = job_receiver.clone();
+                spawn(async move {
+                    loop {
+                        let Ok((input, completion_sender)) = job_receiver.recv().await else {
+                            return;
+                        };
+                        let result = worker_fn(input).await;
+                        if completion_sender.send(result).is_err() {
+                            return;
+                        };
+                    }
+                })
+            })
+            .collect();
+
+        let consumer = spawn(async move {
+            loop {
+                match completion_receiver.recv().await {
+                    Err(_) => {
+                        return;
+                    }
+                    Ok(Completion::Completion(receiver)) => {
+                        let Ok(v) = receiver.await else {
+                            continue;
+                        };
+                        consumer_func(v);
+                    }
+                    Ok(Completion::Sync) => {
+                        if sync_sender.send(()).await.is_err() {
+                            return;
+                        }
+                    }
+                }
+            }
+        });
+
+        Self {
+            job_sender,
+            completion_sender,
+            sync_receiver,
+            workers,
+            consumer,
+        }
+    }
+
+    /// Push a new job to the queue. Blocks until there is space in the queue.
+    pub async fn push_job(&self, job: I) {
+        let (completion_sender, completion_receiver) = oneshot::channel();
+        let _ = self.job_sender.send((job, completion_sender)).await;
+        let _ = self
+            .completion_sender
+            .send(Completion::Completion(completion_receiver))
+            .await;
+    }
+
+    /// Wait for all previously queued jobs to complete.
+    pub async fn flush(&self) {
+        let _ = self.completion_sender.send(Completion::Sync).await;
+        let _ = self.sync_receiver.recv().await;
+    }
+}
+
+impl<I, O> Drop for JobQueue<I, O> {
+    fn drop(&mut self) {
+        self.consumer.abort();
+        for worker in self.workers.drain(..) {
+            worker.abort();
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn test_job_queue() {
+        let result = Arc::new(Mutex::new(Vec::new()));
+        let result_clone = result.clone();
+
+        let job_queue = super::JobQueue::new(
+            6,
+            || Box::new(|i: u32| Box::pin(async move { i })),
+            move |i| result_clone.lock().unwrap().push(i),
+        );
+
+        for i in 0..1000000 {
+            job_queue.push_job(i).await;
+        }
+
+        job_queue.flush().await;
+
+        let expected: Vec<u32> = (0..1000000).collect();
+
+        assert_eq!(&*result.lock().unwrap(), &*expected);
+    }
+}
diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs
@@ -5,7 +5,6 @@ use crate::integrated::delta_table::deletion_vector::{
 };
 use crate::integrated::delta_table::{delta_input_serde_config, register_storage_handlers};
 use crate::transport::{InputEndpoint, InputQueue, InputReaderCommand, IntegratedInputEndpoint};
-use crate::util::JobQueue;
 use crate::{ControllerError, InputConsumer, InputReader, PipelineState};
 use anyhow::{Error as AnyError, Result as AnyResult, anyhow, bail};
 use arrow::array::BooleanArray;
@@ -39,19 +38,20 @@ use deltalake::{DeltaTable, DeltaTableBuilder, Path, datafusion};
 use feldera_adapterlib::format::{ParseError, StagedInputBuffer};
 use feldera_adapterlib::metrics::{ConnectorMetrics, ValueType};
 use feldera_adapterlib::transport::{InputQueueEntry, Resume, Watermark, parse_resume_info};
+use feldera_adapterlib::utils::backoff::calculate_backoff_delay;
 use feldera_adapterlib::utils::datafusion::{
     ColumnNameSet, array_to_string, columns_referenced_by_expression,
     columns_referenced_by_order_by, create_session_context_with, execute_query_collect,
     execute_singleton_query, quote_sql_identifier, timestamp_to_sql_expression,
     validate_sql_expression, validate_sql_order_by, validate_timestamp_column,
 };
+use feldera_adapterlib::utils::job_queue::JobQueue;
 use feldera_storage::tokio::TOKIO_DEDICATED_IO;
 use feldera_types::adapter_stats::ConnectorHealth;
 use feldera_types::config::{FtModel, PipelineConfig};
 use feldera_types::program_schema::{Field, Relation};
 use feldera_types::transport::delta_table::{DeltaTableReaderConfig, DeltaTableTransactionMode};
 use futures_util::StreamExt;
-use rand::Rng;
 use roaring::RoaringTreemap;
 use serde::{Deserialize, Serialize};
 use serde_json::Value as JsonValue;
@@ -72,21 +72,6 @@ use tracing::{debug, info, trace, warn};
 /// Polling interval when following a delta table.
 const POLL_INTERVAL: Duration = Duration::from_millis(1000);
 
-/// Calculate exponential backoff delay for retrying delta log reads.
-/// Starts at 0.5s, doubles each retry, caps at 32s, plus uniform jitter up to 25% of that delay
-/// (capped at `max_delay_ms`) to reduce synchronized retries.
-fn calculate_backoff_delay(retry_count: u32) -> Duration {
-    let base_delay_ms: u64 = 500; // 0.5 seconds
-    let max_delay_ms: u64 = 32_000; // 32 seconds
-    let delay_ms = min(
-        base_delay_ms.checked_shl(retry_count).unwrap_or(u64::MAX),
-        max_delay_ms,
-    );
-    let jitter_span = (delay_ms / 4).max(1);
-    let jitter_ms = rand::thread_rng().gen_range(0..jitter_span);
-    Duration::from_millis(min(delay_ms + jitter_ms, max_delay_ms))
-}
-
 /// Default object store timeout. When not explicitly set by the user,
 /// we use a large timeout value to avoid this issue:
 /// https://github.com/delta-io/delta-rs/issues/2595, which is common for