Kafka input adapter.

Simple adapter that reads data from Kafka into DBSP and doesn't do
anything clever for transactions or reliable delivery.  It is suitable
for streaming data in scenarios where atomic transactions are not
required and at least once delivery is acceptable (i.e., occasional
duplicates can be tolerated).

Author: ryzhyk

Cargo.lock

@@ -16,10 +16,13 @@ once_cell = "1.9.0"
 serde_yaml = "0.9.14"
 csv = { git = "https://github.com/ryzhyk/rust-csv.git" }
 bincode = { version = "2.0.0-rc.2", features = ["serde"] }
+rdkafka = "0.29.0"
 
 [dev-dependencies]
 serde_json = "1.0.87"
 size-of = { version = "0.1.2", features = ["time-std"]}
 tempfile = "3.3.0"
 proptest = "1.0.0"
 proptest-derive = "0.3.0"
+futures = "0.3.25"
+log = "0.4.17"

adapters/Cargo.toml

@@ -863,44 +863,17 @@ impl OutputConsumer for OutputProbe {
 
 #[cfg(test)]
 mod test {
-    use crate::{test::wait, Catalog, Controller, ControllerConfig};
-    use bincode::{Decode, Encode};
+    use crate::{
+        test::{generate_test_data, wait, TestStruct},
+        Catalog, Controller, ControllerConfig,
+    };
     use csv::{ReaderBuilder as CsvReaderBuilder, WriterBuilder as CsvWriterBuilder};
     use dbsp::{DBSPHandle, Runtime};
-    use serde::{Deserialize, Serialize};
     use serde_yaml;
-    use size_of::SizeOf;
     use std::fs::remove_file;
     use tempfile::NamedTempFile;
 
-    use proptest::{collection, prelude::*};
-    use proptest_derive::Arbitrary;
-
-    #[derive(
-        Debug,
-        PartialEq,
-        Eq,
-        PartialOrd,
-        Ord,
-        Serialize,
-        Deserialize,
-        Clone,
-        Hash,
-        SizeOf,
-        Encode,
-        Decode,
-        Arbitrary,
-    )]
-    struct TestStruct {
-        id: u32,
-        b: bool,
-        i: Option<i64>,
-        s: String,
-    }
-
-    fn test_data(size: usize) -> impl Strategy<Value = Vec<TestStruct>> {
-        collection::vec(any::<TestStruct>(), 0..=size)
-    }
+    use proptest::prelude::*;
 
     fn test_circuit(workers: usize) -> (DBSPHandle, Catalog) {
         let (circuit, (input, output)) = Runtime::init_circuit(workers, |circuit| {
@@ -924,7 +897,7 @@ mod test {
         #![proptest_config(ProptestConfig::with_cases(30))]
         #[test]
         fn proptest_csv_file(
-            data in test_data(5000),
+            data in generate_test_data(5000),
             min_batch_size_records in 1..100usize,
             max_buffering_delay_usecs in 1..2000usize,
             input_buffer_size_bytes in 1..1000usize,

adapters/src/controller/mod.rs

@@ -0,0 +1,31 @@
+use bincode::{Decode, Encode};
+use proptest::{collection, prelude::*};
+use proptest_derive::Arbitrary;
+use serde::{Deserialize, Serialize};
+use size_of::SizeOf;
+
+#[derive(
+    Debug,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Serialize,
+    Deserialize,
+    Clone,
+    Hash,
+    SizeOf,
+    Encode,
+    Decode,
+    Arbitrary,
+)]
+pub struct TestStruct {
+    pub id: u32,
+    pub b: bool,
+    pub i: Option<i64>,
+    pub s: String,
+}
+
+pub fn generate_test_data(size: usize) -> impl Strategy<Value = Vec<TestStruct>> {
+    collection::vec(any::<TestStruct>(), 0..=size)
+}

adapters/src/test/data.rs

@@ -8,9 +8,11 @@ use std::{
     time::{Duration, Instant},
 };
 
+mod data;
 mod mock_dezset;
 mod mock_input_consumer;
 
+pub use data::{generate_test_data, TestStruct};
 pub use mock_dezset::MockDeZSet;
 pub use mock_input_consumer::MockInputConsumer;
 

adapters/src/test/mod.rs

@@ -103,7 +103,6 @@ impl FileInputEndpoint {
         status: Arc<AtomicU32>,
         follow: bool,
     ) {
-        println!("file reader worker");
         loop {
             match PipelineState::from_u32(status.load(Ordering::Acquire)) {
                 Some(PipelineState::Paused) => parker.park(),