Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Kafka sink should produce UserErrror when it cannot connect to cluster
Currently if the user misconfigures their Kafka sink, they will get a panic at runtime:

```2023-08-15T03:33:12.267402Z ERROR arroyo_server_common: panicked at 'Producer creation failed: Transaction error: Failed to initialize Producer ID: Local: Timed out

Caused by:
    Failed to initialize Producer ID: Local: Timed out', /opt/arroyo/src/arroyo-worker/src/connectors/kafka/sink/mod.rs:119:14 panic.file="/opt/arroyo/src/arroyo-worker/src/connectors/kafka/sink/mod.rs" panic.line=119 panic.column=14```

This should be handled and produced to the UI as we do in the Kafka source.


Discussion / hints:
Hi @mwylde , I have an interest in arroyo but new to rust. Can I try working on this?
IIUC, the part to send the error to UI for kafka source is [this part](https://github.com/ArroyoSystems/arroyo/blob/df3360980f6681d690d0205382479d3589ed3a1f/arroyo-worker/src/connectors/kafka/source/mod.rs#L205-L210)?

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index fe44a6fb3..3a89a4751 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -102,9 +102,9 @@ dependencies = [
 
 [[package]]
 name = "anstream"
-version = "0.6.5"
+version = "0.6.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d664a92ecae85fd0a7392615844904654d1d5f5514837f471ddef4a057aba1b6"
+checksum = "6e2e1ebcb11de5c03c67de28a7df593d32191b44939c482e97702baaaa6ab6a5"
 dependencies = [
  "anstyle",
  "anstyle-parse",
@@ -370,7 +370,7 @@ dependencies = [
  "arrow-data 49.0.0",
  "arrow-schema 49.0.0",
  "arrow-select 49.0.0",
- "base64 0.21.5",
+ "base64 0.21.7",
  "chrono",
  "comfy-table",
  "half",
@@ -491,7 +491,7 @@ dependencies = [
 [[package]]
 name = "arrow-json"
 version = "49.0.0"
-source = "git+https://github.com/ArroyoSystems/arrow-rs?branch=49.0.0/json#bc9aa2e6e69e50b9eb8709e950426f7351b9c3ac"
+source = "git+https://github.com/ArroyoSystems/arrow-rs?branch=49.0.0/json#d5ffe4eb068a5ea5904b86caf47e4a22929768f4"
 dependencies = [
  "arrow-array 49.0.0",
  "arrow-buffer 49.0.0",
@@ -646,7 +646,7 @@ version = "0.9.0-dev"
 dependencies = [
  "anyhow",
  "bollard",
- "clap 4.4.12",
+ "clap 4.4.18",
  "open",
  "reqwest",
  "tokio",
@@ -673,7 +673,7 @@ dependencies = [
  "async-trait",
  "axum",
  "axum-extra",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bincode 2.0.0-rc.3",
  "chrono",
  "cornucopia",
@@ -698,7 +698,7 @@ dependencies = [
  "schemars",
  "serde",
  "serde_json",
- "syn 2.0.46",
+ "syn 2.0.48",
  "thiserror",
  "time",
  "tokio",
@@ -712,7 +712,7 @@ dependencies = [
  "tracing",
  "typify",
  "utoipa 3.5.0",
- "utoipa 4.1.0",
+ "utoipa 4.2.0",
  "utoipa-swagger-ui",
  "uuid",
 ]
@@ -778,7 +778,7 @@ dependencies = [
  "arroyo-types",
  "async-stream",
  "async-trait",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bincode 2.0.0-rc.3",
  "chrono",
  "cornucopia",
@@ -802,7 +802,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_yaml",
- "syn 2.0.46",
+ "syn 2.0.48",
  "thiserror",
  "time",
  "tokio",
@@ -837,7 +837,7 @@ dependencies = [
  "serde",
  "serde_json",
  "strum",
- "syn 2.0.46",
+ "syn 2.0.48",
  "tokio",
  "toml 0.7.8",
  "tonic",
@@ -875,9 +875,10 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_json_path",
- "syn 2.0.46",
+ "syn 2.0.48",
  "tokio",
  "tokio-stream",
+ "toml 0.8.8",
  "tracing",
  "typify",
  "unicase",
@@ -911,7 +912,7 @@ version = "0.9.0-dev"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -954,7 +955,7 @@ dependencies = [
  "serde_json",
  "serde_with",
  "url",
- "utoipa 4.1.0",
+ "utoipa 4.2.0",
 ]
 
 [[package]]
@@ -967,7 +968,7 @@ dependencies = [
  "arrow-schema 49.0.0",
  "arroyo-types",
  "async-trait",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bincode 2.0.0-rc.3",
  "log",
  "nanoid",
@@ -980,7 +981,7 @@ dependencies = [
  "tonic",
  "tonic-build",
  "tracing",
- "utoipa 4.1.0",
+ "utoipa 4.2.0",
 ]
 
 [[package]]
@@ -1032,7 +1033,7 @@ dependencies = [
  "quote",
  "serde",
  "serde_json",
- "syn 2.0.46",
+ "syn 2.0.48",
  "test-log",
  "tokio",
  "tracing",
@@ -1105,6 +1106,7 @@ version = "0.9.0-dev"
 dependencies = [
  "arrow 49.0.0",
  "arrow-array 49.0.0",
+ "async-trait",
  "bincode 2.0.0-rc.3",
  "chrono",
  "serde",
@@ -1136,7 +1138,7 @@ dependencies = [
  "async-trait",
  "aws-config",
  "aws-sdk-kinesis",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bincode 2.0.0-rc.3",
  "bytes",
  "chrono",
@@ -1210,7 +1212,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ca33f4bc4ed1babef42cad36cc1f51fa88be00420404e5b1e80ab1b18f7678c"
 dependencies = [
  "concurrent-queue",
- "event-listener 4.0.2",
+ "event-listener 4.0.3",
  "event-listener-strategy",
  "futures-core",
  "pin-project-lite",
@@ -1218,9 +1220,9 @@ dependencies = [
 
 [[package]]
 name = "async-compression"
-version = "0.4.5"
+version = "0.4.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc2d0cfb2a7388d34f590e76686704c494ed7aaceed62ee1ba35cbf363abc2a5"
+checksum = "a116f46a969224200a0a97f29cfd4c50e7534e4b4826bd23ea2c3c533039c82c"
 dependencies = [
  "bzip2",
  "flate2",
@@ -1240,11 +1242,11 @@ version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "17ae5ebefcc48e7452b4987947920dac9450be1110cadf34d1b8c116bdbaf97c"
 dependencies = [
- "async-lock 3.2.0",
+ "async-lock 3.3.0",
  "async-task",
  "concurrent-queue",
  "fastrand 2.0.1",
- "futures-lite 2.1.0",
+ "futures-lite 2.2.0",
  "slab",
 ]
 
@@ -1256,10 +1258,10 @@ checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c"
 dependencies = [
  "async-channel 2.1.1",
  "async-executor",
- "async-io 2.2.2",
- "async-lock 3.2.0",
+ "async-io 2.3.0",
+ "async-lock 3.3.0",
  "blocking",
- "futures-lite 2.1.0",
+ "futures-lite 2.2.0",
  "once_cell",
 ]
 
@@ -1285,18 +1287,18 @@ dependencies = [
 
 [[package]]
 name = "async-io"
-version = "2.2.2"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6afaa937395a620e33dc6a742c593c01aced20aa376ffb0f628121198578ccc7"
+checksum = "fb41eb19024a91746eba0773aa5e16036045bbf45733766661099e182ea6a744"
 dependencies = [
- "async-lock 3.2.0",
+ "async-lock 3.3.0",
  "cfg-if",
  "concurrent-queue",
  "futures-io",
- "futures-lite 2.1.0",
+ "futures-lite 2.2.0",
  "parking",
- "polling 3.3.1",
- "rustix 0.38.28",
+ "polling 3.3.2",
+ "rustix 0.38.30",
  "slab",
  "tracing",
  "windows-sys 0.52.0",
@@ -1313,11 +1315,11 @@ dependencies = [
 
 [[package]]
 name = "async-lock"
-version = "3.2.0"
+version = "3.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7125e42787d53db9dd54261812ef17e937c95a51e4d291373b670342fa44310c"
+checksum = "d034b430882f8381900d3fe6f0aaa3ad94f2cb4ac519b429692a1bc2dda4ae7b"
 dependencies = [
- "event-listener 4.0.2",
+ "event-listener 4.0.3",
  "event-listener-strategy",
  "pin-project-lite",
 ]
@@ -1355,7 +1357,7 @@ dependencies = [
  "cfg-if",
  "event-listener 3.1.0",
  "futures-lite 1.13.0",
- "rustix 0.38.28",
+ "rustix 0.38.30",
  "windows-sys 0.48.0",
 ]
 
@@ -1375,13 +1377,13 @@ version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9e47d90f65a225c4527103a8d747001fc56e375203592b25ad103e1ca13124c5"
 dependencies = [
- "async-io 2.2.2",
+ "async-io 2.3.0",
  "async-lock 2.8.0",
  "atomic-waker",
  "cfg-if",
  "futures-core",
  "futures-io",
- "rustix 0.38.28",
+ "rustix 0.38.30",
  "signal-hook-registry",
  "slab",
  "windows-sys 0.48.0",
@@ -1433,14 +1435,14 @@ checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
 name = "async-task"
-version = "4.6.0"
+version = "4.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e1d90cd0b264dfdd8eb5bad0a2c217c1f88fa96a8573f40e7b12de23fb468f46"
+checksum = "fbb36e985947064623dbd357f727af08ffd077f93d696782f3c56365fa2e2799"
 
 [[package]]
 name = "async-trait"
@@ -1450,7 +1452,7 @@ checksum = "c980ee35e870bd1a4d2c8294d4c04d0499e67bca1e4b5cefcc693c2fa00caea9"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -1852,7 +1854,7 @@ dependencies = [
  "heck",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -1910,9 +1912,9 @@ checksum = "0ea22880d78093b0cbe17c89f64a7d457941e65759157ec6cb31a31d652b05e5"
 
 [[package]]
 name = "base64"
-version = "0.21.5"
+version = "0.21.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "35636a1494ede3b646cc98f74f8e62c773a38a659ebc777a2cf26b9b74171df9"
+checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567"
 
 [[package]]
 name = "base64ct"
@@ -1962,9 +1964,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
 [[package]]
 name = "bitflags"
-version = "2.4.1"
+version = "2.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "327762f6e5a765692301e5bb513e0d9fef63be86bbc14528052b1cd3e6f03e07"
+checksum = "ed570934406eb16438a4e976b1b4500774099c13b8cb96eec99f620f05090ddf"
 
 [[package]]
 name = "blake2"
@@ -2013,11 +2015,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6a37913e8dc4ddcc604f0c6d3bf2887c995153af3611de9e23c352b44c1b9118"
 dependencies = [
  "async-channel 2.1.1",
- "async-lock 3.2.0",
+ "async-lock 3.3.0",
  "async-task",
  "fastrand 2.0.1",
  "futures-io",
- "futures-lite 2.1.0",
+ "futures-lite 2.2.0",
  "piper",
  "tracing",
 ]
@@ -2028,7 +2030,7 @@ version = "0.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f03db470b3c0213c47e978da93200259a1eb4dae2e5512cba9955e2b540a6fc6"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
  "bollard-stubs",
  "bytes",
  "futures-core",
@@ -2147,6 +2149,38 @@ dependencies = [
  "pkg-config",
 ]
 
+[[package]]
+name = "camino"
+version = "1.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c59e92b5a388f549b863a7bea62612c09f24c8393560709a54558a9abdfb3b9c"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "cargo-platform"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ceed8ef69d8518a5dda55c07425450b58a4e1946f4951eab6d7191ee86c2443d"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "cargo_metadata"
+version = "0.18.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d886547e41f740c616ae73108f6eb70afe6d940c7bc697cb30f13daec073037"
+dependencies = [
+ "camino",
+ "cargo-platform",
+ "semver",
+ "serde",
+ "serde_json",
+ "thiserror",
+]
+
 [[package]]
 name = "cc"
 version = "1.0.83"
@@ -2228,9 +2262,9 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "4.4.12"
+version = "4.4.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dcfab8ba68f3668e89f6ff60f5b205cea56aa7b769451a59f34b8682f51c056d"
+checksum = "1e578d6ec4194633722ccf9544794b71b1385c3c027efe0c55db226fc880865c"
 dependencies = [
  "clap_builder",
  "clap_derive 4.4.7",
@@ -2238,9 +2272,9 @@ dependencies = [
 
 [[package]]
 name = "clap_builder"
-version = "4.4.12"
+version = "4.4.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb7fb5e4e979aec3be7791562fcba452f94ad85e954da024396433e0e25a79e9"
+checksum = "4df4df40ec50c46000231c914968278b1eb05098cf8f1b3a518a95030e71d1c7"
 dependencies = [
  "anstream",
  "anstyle",
@@ -2270,7 +2304,7 @@ dependencies = [
  "heck",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -2442,7 +2476,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3d21d8c5cf5940c13b1d82977fc5d4230714ce76d991c5c00b8a1c791c691dcf"
 dependencies = [
  "chumsky",
- "clap 4.4.12",
+ "clap 4.4.18",
  "codegen_template",
  "heck",
  "indexmap 1.9.3",
@@ -2497,9 +2531,9 @@ dependencies = [
 
 [[package]]
 name = "cpufeatures"
-version = "0.2.11"
+version = "0.2.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce420fe07aecd3e67c5f910618fe65e94158f6dcc0adf44e00d69ce2bdfe0fd0"
+checksum = "53fe5e26ff1b7aef8bca9c6080520cfb8d9333c7568e1829cef191a9723e5504"
 dependencies = [
  "libc",
 ]
@@ -2638,44 +2672,37 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-channel"
-version = "0.5.10"
+version = "0.5.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "82a9b73a36529d9c47029b9fb3a6f0ea3cc916a261195352ba19e770fc1748b2"
+checksum = "176dc175b78f56c0f321911d9c8eb2b77a78a4860b9c19db83835fea1a46649b"
 dependencies = [
- "cfg-if",
  "crossbeam-utils",
 ]
 
 [[package]]
 name = "crossbeam-deque"
-version = "0.8.4"
+version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fca89a0e215bab21874660c67903c5f143333cab1da83d041c7ded6053774751"
+checksum = "613f8cc01fe9cf1a3eb3d7f488fd2fa8388403e97039e2f73692932e291a770d"
 dependencies = [
- "cfg-if",
  "crossbeam-epoch",
  "crossbeam-utils",
 ]
 
 [[package]]
 name = "crossbeam-epoch"
-version = "0.9.17"
+version = "0.9.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e3681d554572a651dda4186cd47240627c3d0114d45a95f6ad27f2f22e7548d"
+checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
 dependencies = [
- "autocfg",
- "cfg-if",
  "crossbeam-utils",
 ]
 
 [[package]]
 name = "crossbeam-utils"
-version = "0.8.18"
+version = "0.8.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3a430a770ebd84726f584a90ee7f020d28db52c6d02138900f22341f866d39c"
-dependencies = [
- "cfg-if",
-]
+checksum = "248e3bacc7dc6baa3b21e405ee045c3047101a49145e7e9eca583ab4c2ca5345"
 
 [[package]]
 name = "crunchy"
@@ -2758,7 +2785,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "30d2b3721e861707777e3195b0158f950ae6dc4a27e4d02ff9f67e3eb3de199e"
 dependencies = [
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -2806,7 +2833,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "strsim",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -2828,7 +2855,7 @@ checksum = "836a9bbc7ad63342d6d6e7b815ccab164bc77a2d95d84bc3117a8c0d5c98e2d5"
 dependencies = [
  "darling_core 0.20.3",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -2990,7 +3017,7 @@ dependencies = [
  "arrow-buffer 49.0.0",
  "arrow-ord 49.0.0",
  "arrow-schema 49.0.0",
- "base64 0.21.5",
+ "base64 0.21.7",
  "blake2",
  "blake3",
  "chrono",
@@ -3431,14 +3458,14 @@ dependencies = [
  "num-traits",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
 name = "env_logger"
-version = "0.10.1"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "95b3f3e67048839cb0d0781f445682a35113da7121f7c949db0e2be96a4fbece"
+checksum = "4cd405aab171cb85d6735e5c8d9db038c17d3ca007a4d2c25f337935c3d90580"
 dependencies = [
  "humantime",
  "is-terminal",
@@ -3482,9 +3509,9 @@ dependencies = [
 
 [[package]]
 name = "event-listener"
-version = "4.0.2"
+version = "4.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "218a870470cce1469024e9fb66b901aa983929d81304a1cdb299f28118e550d5"
+checksum = "67b215c49b2b248c855fb73579eb1f4f26c38ffdc12973e20e07b91d78d5646e"
 dependencies = [
  "concurrent-queue",
  "parking",
@@ -3497,7 +3524,7 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "958e4d70b6d5e81971bebec42271ec641e7ff4e170a6fa605f2b8a8b65cb97d3"
 dependencies = [
- "event-listener 4.0.2",
+ "event-listener 4.0.3",
  "pin-project-lite",
 ]
 
@@ -3614,16 +3641,16 @@ dependencies = [
 
 [[package]]
 name = "fluvio"
-version = "0.21.3"
+version = "0.21.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c4117a72f05d65f1ca17fe3b846162a5cb1175f1446570040734ebb3360dc8bf"
+checksum = "c11dbda79ac9b02000d961e5b1296148f82800c200890d7c9732d2b0094e51ba"
 dependencies = [
  "anyhow",
  "async-channel 1.9.0",
  "async-lock 2.8.0",
  "async-rwlock",
  "async-trait",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "cfg-if",
  "chrono",
@@ -3668,20 +3695,20 @@ dependencies = [
 
 [[package]]
 name = "fluvio-controlplane-metadata"
-version = "0.25.0"
+version = "0.25.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b8328c851ecad841923904cdbf39a9adfef1f2d35e3f79e4c1563d6e9f9efcad"
+checksum = "9ba2faa10cf48b39fc8140f9d0b1abf3866f5c8461dbe666f475627ce2e53a49"
 dependencies = [
  "anyhow",
  "async-trait",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "bytesize",
  "derive_builder",
  "flate2",
  "fluvio-future",
  "fluvio-protocol",
- "fluvio-stream-model",
+ "fluvio-stream-model 0.11.1",
  "fluvio-types",
  "flv-util",
  "humantime-serde",
@@ -3721,9 +3748,9 @@ dependencies = [
 
 [[package]]
 name = "fluvio-protocol"
-version = "0.10.7"
+version = "0.10.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70228d5c0d411c218887400d140658cb4430a18a91a633b88e87d1886eb5800e"
+checksum = "e59e488ff6f3e24e5ec654e62c7bc58e8e26688fbc20298d03eaa0b1b0dca268"
 dependencies = [
  "bytes",
  "content_inspector",
@@ -3763,7 +3790,7 @@ dependencies = [
  "fluvio-controlplane-metadata",
  "fluvio-protocol",
  "fluvio-socket",
- "fluvio-stream-model",
+ "fluvio-stream-model 0.10.0",
  "fluvio-types",
  "paste",
  "static_assertions",
@@ -3792,14 +3819,14 @@ checksum = "71466f33efdca5973620ca6de3ad69c17e98b6fdf2c40cf52bb0fae6fd3e21ba"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
 name = "fluvio-socket"
-version = "0.14.6"
+version = "0.14.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0a46d41a5958604b7c05e340c500acde224e03a845dd4eb2aad3e2ed9ebf9e18"
+checksum = "ced324ccb056bf353a742b29bbae460d5d9199a817d57ba3c75b2b42c00cfe31"
 dependencies = [
  "async-channel 1.9.0",
  "async-lock 2.8.0",
@@ -3854,6 +3881,20 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "fluvio-stream-model"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ef7b57909ffd437f974238f692b6e10f3e3c8af72caab90a8d44865d2e744cb2"
+dependencies = [
+ "async-rwlock",
+ "event-listener 3.1.0",
+ "k8-types",
+ "once_cell",
+ "serde",
+ "tracing",
+]
+
 [[package]]
 name = "fluvio-types"
 version = "0.4.4"
@@ -3994,9 +4035,9 @@ dependencies = [
 
 [[package]]
 name = "futures-lite"
-version = "2.1.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aeee267a1883f7ebef3700f262d2d54de95dfaf38189015a74fdc4e0c7ad8143"
+checksum = "445ba825b27408685aaecefd65178908c36c6e96aaf6d8599419d46e624192ba"
 dependencies = [
  "fastrand 2.0.1",
  "futures-core",
@@ -4013,7 +4054,7 @@ checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -4067,7 +4108,7 @@ version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "27d12c0aed7f1e24276a241aadc4cb8ea9f83000f34bc062b7cc2d51e3b0fabd"
 dependencies = [
- "bitflags 2.4.1",
+ "bitflags 2.4.2",
  "debugid",
  "fxhash",
  "serde",
@@ -4087,9 +4128,9 @@ dependencies = [
 
 [[package]]
 name = "getrandom"
-version = "0.2.11"
+version = "0.2.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fe9006bed769170c11f845cf00c7c1e9092aeb3f268e007c3e760ac68008070f"
+checksum = "190092ea657667030ac6a35e305e62fc4dd69fd98ac98631e5d3a2b1575a12b5"
 dependencies = [
  "cfg-if",
  "js-sys",
@@ -4164,9 +4205,9 @@ dependencies = [
 
 [[package]]
 name = "h2"
-version = "0.3.22"
+version = "0.3.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d6250322ef6e60f93f9a2162799302cd6f68f79f6e5d85c8c16f14d1d958178"
+checksum = "bb2c4422095b67ee78da96fbb51a4cc413b3b25883c7717ff7ca1ab31022c9c9"
 dependencies = [
  "bytes",
  "fnv",
@@ -4223,7 +4264,7 @@ version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06683b93020a07e3dbcf5f8c0f6d40080d725bea7936fc01ad345c01b97dc270"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "headers-core",
  "http",
@@ -4258,9 +4299,9 @@ dependencies = [
 
 [[package]]
 name = "hermit-abi"
-version = "0.3.3"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d77f7ec81a6d05a3abb01ab6eb7590f6083d08449fe5a1c8b1e620283546ccb7"
+checksum = "5d3d0e0f38255e7fa3cf31335b3a56f05febd18025f4db5ef7a0cfb4f8da651f"
 
 [[package]]
 name = "hex"
@@ -4617,7 +4658,7 @@ version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eae7b9aee968036d54dce06cebaefd919e4472e753296daccd6d344e3e2df0c2"
 dependencies = [
- "hermit-abi 0.3.3",
+ "hermit-abi 0.3.4",
  "libc",
  "windows-sys 0.48.0",
 ]
@@ -4643,8 +4684,8 @@ version = "0.4.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0bad00257d07be169d870ab665980b06cdb366d792ad690bf2e76876dc503455"
 dependencies = [
- "hermit-abi 0.3.3",
- "rustix 0.38.28",
+ "hermit-abi 0.3.4",
+ "rustix 0.38.30",
  "windows-sys 0.52.0",
 ]
 
@@ -4728,9 +4769,9 @@ dependencies = [
 
 [[package]]
 name = "js-sys"
-version = "0.3.66"
+version = "0.3.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cee9c64da59eae3b50095c18d3e74f8b73c0b86d2792824ff01bbce68ba229ca"
+checksum = "9a1d36f1235bc969acba30b7f5990b864423a6068a10f7c90ae8f0112e3a59d1"
 dependencies = [
  "wasm-bindgen",
 ]
@@ -4792,9 +4833,9 @@ dependencies = [
 
 [[package]]
 name = "k256"
-version = "0.13.2"
+version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f01b677d82ef7a676aa37e099defd83a28e15687112cafdd112d60236b6115b"
+checksum = "956ff9b67e26e1a6a866cb758f12c6f8746208489e3e4a4b5580802f2f0a587b"
 dependencies = [
  "cfg-if",
  "ecdsa",
@@ -4820,7 +4861,7 @@ version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cd990069640f9db34b3b0f7a1afc62a05ffaa3be9b66aa3c313f58346df7f788"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "chrono",
  "http",
@@ -5054,9 +5095,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.151"
+version = "0.2.152"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "302d7ab3130588088d277783b1e2d2e10c9e9e4a16dd9050e6ec93fb3e7048f4"
+checksum = "13e3bf6590cbc649f4d1a3eefc9d5d6eb746f5200ffb04e5e142700b8faa56e7"
 
 [[package]]
 name = "libflate"
@@ -5114,16 +5155,16 @@ version = "0.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85c833ca1e66078851dba29046874e38f08b2c883700aa29a03ddd3b23814ee8"
 dependencies = [
- "bitflags 2.4.1",
+ "bitflags 2.4.2",
  "libc",
  "redox_syscall 0.4.1",
 ]
 
 [[package]]
 name = "libz-sys"
-version = "1.1.12"
+version = "1.1.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d97137b25e321a73eef1418d1d5d2eda4d77e12813f8e6dead84bc52c5870a7b"
+checksum = "295c17e837573c8c821dbaeb3cceb3d745ad082f7572191409e69cbc1b3fd050"
 dependencies = [
  "cc",
  "libc",
@@ -5160,9 +5201,9 @@ checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519"
 
 [[package]]
 name = "linux-raw-sys"
-version = "0.4.12"
+version = "0.4.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c4cd1a83af159aa67994778be9070f0ae1bd732942279cabb14f86f986a21456"
+checksum = "01cda141df6706de531b6c46c3a33ecca755538219bd484262fa09410c13539c"
 
 [[package]]
 name = "local-ip-address"
@@ -5217,9 +5258,9 @@ dependencies = [
 
 [[package]]
 name = "lz4_flex"
-version = "0.11.1"
+version = "0.11.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3ea9b256699eda7b0387ffbc776dd625e28bde3918446381781245b7a50349d8"
+checksum = "912b45c753ff5f7f5208307e8ace7d2a2e30d024e26d3509f3dce546c044ce15"
 dependencies = [
  "twox-hash",
 ]
@@ -5307,7 +5348,7 @@ version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b2cffa4ad52c6f791f4f8b15f0c05f9824b2ced1160e88cc393d64fff9a8ac64"
 dependencies = [
- "rustix 0.38.28",
+ "rustix 0.38.30",
 ]
 
 [[package]]
@@ -5357,7 +5398,7 @@ checksum = "49e7bc1560b95a3c4a25d03de42fe76ca718ab92d1a22a55b9b4cf67b3ae635c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -5487,7 +5528,7 @@ version = "0.27.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2eb04e9c688eff1c89d72b407f168cf79bb9e867a9d3323ed6c01519eb9cc053"
 dependencies = [
- "bitflags 2.4.1",
+ "bitflags 2.4.2",
  "cfg-if",
  "libc",
 ]
@@ -5624,7 +5665,7 @@ version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43"
 dependencies = [
- "hermit-abi 0.3.3",
+ "hermit-abi 0.3.4",
  "libc",
 ]
 
@@ -5686,7 +5727,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f930c88a43b1c3f6e776dfe495b4afab89882dbc81530c632db2ed65451ebcb4"
 dependencies = [
  "async-trait",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "chrono",
  "futures",
@@ -5714,7 +5755,7 @@ version = "0.8.0"
 source = "git+https://github.com/ArroyoSystems/arrow-rs?branch=0.8.0/put_part_api#cdf7298b0d4b801d2d55d7436d766714bed7e297"
 dependencies = [
  "async-trait",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "chrono",
  "futures",
@@ -5766,7 +5807,7 @@ version = "0.10.62"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8cde4d2d9200ad5909f8dac647e29482e07c3a35de8a13fce7c9c7747ad9f671"
 dependencies = [
- "bitflags 2.4.1",
+ "bitflags 2.4.2",
  "cfg-if",
  "foreign-types",
  "libc",
@@ -5783,7 +5824,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -5958,7 +5999,7 @@ dependencies = [
  "arrow-ipc 46.0.0",
  "arrow-schema 46.0.0",
  "arrow-select 46.0.0",
- "base64 0.21.5",
+ "base64 0.21.7",
  "brotli",
  "bytes",
  "chrono",
@@ -5991,7 +6032,7 @@ dependencies = [
  "arrow-ipc 49.0.0",
  "arrow-schema 49.0.0",
  "arrow-select 49.0.0",
- "base64 0.21.5",
+ "base64 0.21.7",
  "brotli",
  "bytes",
  "chrono",
@@ -6153,7 +6194,7 @@ checksum = "4359fd9c9171ec6e8c62926d6faaf553a8dc3f64e1507e76da7911b4f6a04405"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -6213,9 +6254,9 @@ dependencies = [
 
 [[package]]
 name = "pkg-config"
-version = "0.3.28"
+version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69d3587f8a9e599cc7ec2c00e331f71c4e69a5f9a4b8a6efd5b07466b9736f9a"
+checksum = "2900ede94e305130c13ddd391e0ab7cbaeb783945ae07a279c268cb05109c6cb"
 
 [[package]]
 name = "polling"
@@ -6235,14 +6276,14 @@ dependencies = [
 
 [[package]]
 name = "polling"
-version = "3.3.1"
+version = "3.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cf63fa624ab313c11656b4cda960bfc46c410187ad493c41f6ba2d8c1e991c9e"
+checksum = "545c980a3880efd47b2e262f6a4bb6daad6555cf3367aa9c4e52895f69537a41"
 dependencies = [
  "cfg-if",
  "concurrent-queue",
  "pin-project-lite",
- "rustix 0.38.28",
+ "rustix 0.38.30",
  "tracing",
  "windows-sys 0.52.0",
 ]
@@ -6270,7 +6311,7 @@ dependencies = [
  "heck",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -6279,7 +6320,7 @@ version = "0.6.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "49b6c5ef183cd3ab4ba005f1ca64c21e8bd97ce4699cfea9e8d9a2c4958ca520"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
  "byteorder",
  "bytes",
  "fallible-iterator",
@@ -6346,7 +6387,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a41cf62165e97c7f814d2221421dbb9afcbcdb0a88068e5ea206e19951c2cbb5"
 dependencies = [
  "proc-macro2",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -6394,9 +6435,9 @@ dependencies = [
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.74"
+version = "1.0.76"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2de98502f212cfcea8d0bb305bd0f49d7ebdd75b64ba0a68f937d888f4e0d6db"
+checksum = "95fc56cda0b5c3325f5fbbd7ff9fda9e02bb00bb3dac51252d2f1bfa1cb8cc8c"
 dependencies = [
  "unicode-ident",
 ]
@@ -6483,7 +6524,7 @@ dependencies = [
  "prost 0.12.3",
  "prost-types",
  "regex",
- "syn 2.0.46",
+ "syn 2.0.48",
  "tempfile",
  "which",
 ]
@@ -6511,7 +6552,7 @@ dependencies = [
  "itertools 0.11.0",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -6671,9 +6712,9 @@ dependencies = [
 
 [[package]]
 name = "rayon"
-version = "1.8.0"
+version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c27db03db7734835b3f53954b534c91069375ce6ccaa2e065441e07d9b6cdb1"
+checksum = "fa7237101a77a10773db45d62004a272517633fbcc3df19d96455ede1122e051"
 dependencies = [
  "either",
  "rayon-core",
@@ -6681,9 +6722,9 @@ dependencies = [
 
 [[package]]
 name = "rayon-core"
-version = "1.12.0"
+version = "1.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5ce3fb6ad83f861aac485e76e1985cd109d9a3713802152be56c3b1f0e0658ed"
+checksum = "1465873a3dfdaa8ae7cb14b4383657caab0b3e8a0aa9ae8e04b044854c8dfce2"
 dependencies = [
  "crossbeam-deque",
  "crossbeam-utils",
@@ -6823,7 +6864,7 @@ dependencies = [
  "quote",
  "refinery-core",
  "regex",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -6911,7 +6952,7 @@ version = "0.11.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "37b1ae8d9ac08420c66222fb9096fc5de435c3c48542bc5336c51892cffafb41"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "encoding_rs",
  "futures-core",
@@ -7223,7 +7264,7 @@ dependencies = [
  "quote",
  "rust-embed-utils",
  "shellexpand",
- "syn 2.0.46",
+ "syn 2.0.48",
  "walkdir",
 ]
 
@@ -7288,14 +7329,14 @@ dependencies = [
 
 [[package]]
 name = "rustix"
-version = "0.38.28"
+version = "0.38.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72e572a5e8ca657d7366229cdde4bd14c4eb5499a9573d4d366fe1b599daa316"
+checksum = "322394588aaf33c24007e8bb3238ee3e4c5c09c084ab32bc73890b99ff326bca"
 dependencies = [
- "bitflags 2.4.1",
+ "bitflags 2.4.2",
  "errno",
  "libc",
- "linux-raw-sys 0.4.12",
+ "linux-raw-sys 0.4.13",
  "windows-sys 0.52.0",
 ]
 
@@ -7354,7 +7395,7 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1c74cae0a4cf6ccbbf5f359f08efdf8ee7e1dc532573bf0db71968cb56b1448c"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
 ]
 
 [[package]]
@@ -7529,9 +7570,9 @@ checksum = "a3f0bf26fd526d2a95683cd0f87bf103b8539e2ca1ef48ce002d67aad59aa0b4"
 
 [[package]]
 name = "serde"
-version = "1.0.194"
+version = "1.0.195"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b114498256798c94a0689e1a15fec6005dee8ac1f41de56404b67afc2a4b773"
+checksum = "63261df402c67811e9ac6def069e4786148c4563f4b50fd4bf30aa370d626b02"
 dependencies = [
  "serde_derive",
 ]
@@ -7548,13 +7589,13 @@ dependencies = [
 
 [[package]]
 name = "serde_derive"
-version = "1.0.194"
+version = "1.0.195"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a3385e45322e8f9931410f01b3031ec534c3947d0e94c18049af4d9f9907d4e0"
+checksum = "46fe8f8603d81ba86327b23a2e9cdf49e1255fb94a4c5f297f6ee0547178ea2c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -7570,9 +7611,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.110"
+version = "1.0.111"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6fbd975230bada99c8bb618e0c365c2eefa219158d5c6c29610fd09ff1833257"
+checksum = "176e46fa42316f18edd598015a5166857fc835ec732f5215eac6b7bdbf0a84f4"
 dependencies = [
  "indexmap 2.1.0",
  "itoa",
@@ -7630,7 +7671,7 @@ checksum = "bb9387330da43020c17237e22c76bd19c93305c75d99ec962c58f385c7e1f5ad"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -7651,7 +7692,7 @@ checksum = "0b2e6b945e9d3df726b65d6ee24060aff8e3533d431f677a9695db04eff9dfdb"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -7672,7 +7713,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "serde",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -7693,7 +7734,7 @@ version = "3.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "64cd236ccc1b7a29e7e2739f27c0b2dd199804abc4290e32f59f3b68d6405c23"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
  "chrono",
  "hex",
  "indexmap 1.9.3",
@@ -7713,7 +7754,7 @@ dependencies = [
  "darling 0.20.3",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -7862,9 +7903,9 @@ checksum = "826167069c09b99d56f31e9ae5c99049e932a98c9dc2dac47645b08dbbf76ba7"
 
 [[package]]
 name = "smallvec"
-version = "1.11.2"
+version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4dccd0940a2dcdf68d092b8cbab7dc0ad8fa938bf95787e1b916b0e3d0e8e970"
+checksum = "e6ecd384b10a64542d77071bd64bd7b231f4ed5940fba55e98c3de13824cf3d7"
 
 [[package]]
 name = "smawk"
@@ -7976,7 +8017,7 @@ checksum = "01b2e185515564f15375f593fb966b5718bc624ba77fe49fa4616ad619690554"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8040,7 +8081,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8113,9 +8154,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.46"
+version = "2.0.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89456b690ff72fddcecf231caedbe615c59480c93358a93dfae7fc29e3ebbf0e"
+checksum = "0f3531638e407dfc0814761abb7c00a5b54992b849452a0646b7f65c9f770f3f"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -8164,15 +8205,15 @@ dependencies = [
  "cfg-if",
  "fastrand 2.0.1",
  "redox_syscall 0.4.1",
- "rustix 0.38.28",
+ "rustix 0.38.30",
  "windows-sys 0.52.0",
 ]
 
 [[package]]
 name = "termcolor"
-version = "1.4.0"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ff1bc3d3f05aff0403e8ac0d92ced918ec05b666a43f83297ccef5bea8a3d449"
+checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755"
 dependencies = [
  "winapi-util",
 ]
@@ -8205,7 +8246,7 @@ dependencies = [
  "cfg-if",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8216,7 +8257,7 @@ checksum = "5c89e72a01ed4c579669add59014b9a524d609c0c88c6a585ce37485879f6ffb"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
  "test-case-core",
 ]
 
@@ -8238,7 +8279,7 @@ checksum = "7ba277e77219e9eea169e8508942db1bf5d8a41ff2db9b20aab5a5aadc9fa25d"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8275,7 +8316,7 @@ checksum = "fa0faa943b50f3db30a20aa7e265dbc66076993efed8463e8de414e5d06d3471"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8392,7 +8433,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8595,7 +8636,7 @@ dependencies = [
  "async-stream",
  "async-trait",
  "axum",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "h2",
  "http",
@@ -8623,7 +8664,7 @@ dependencies = [
  "proc-macro2",
  "prost-build",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8645,7 +8686,7 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0fddb2a37b247e6adcb9f239f4e5cefdcc5ed526141a416b943929f13aea2cce"
 dependencies = [
- "base64 0.21.5",
+ "base64 0.21.7",
  "bytes",
  "http",
  "http-body",
@@ -8685,8 +8726,8 @@ version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61c5bb1d698276a2443e5ecfabc1008bf15a36c12e6a7176e7bf089ea9131140"
 dependencies = [
- "base64 0.21.5",
- "bitflags 2.4.1",
+ "base64 0.21.7",
+ "bitflags 2.4.2",
  "bytes",
  "futures-core",
  "futures-util",
@@ -8749,7 +8790,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8865,7 +8906,7 @@ checksum = "f03ca4cb38206e2bef0700092660bb74d696f808514dae47fa1467cbfe26e96e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -8895,7 +8936,7 @@ dependencies = [
  "regress",
  "schemars",
  "serde_json",
- "syn 2.0.46",
+ "syn 2.0.48",
  "thiserror",
  "unicode-ident",
 ]
@@ -8911,7 +8952,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_tokenstream",
- "syn 2.0.46",
+ "syn 2.0.48",
  "typify-impl",
 ]
 
@@ -8926,9 +8967,9 @@ dependencies = [
 
 [[package]]
 name = "unicode-bidi"
-version = "0.3.14"
+version = "0.3.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f2528f27a9eb2b21e69c95319b30bd0efd85d09c379741b0f78ea1d86be2416"
+checksum = "08f95100a766bf4f8f28f90d77e0a5461bbdb219042e7679bebe79004fed8d75"
 
 [[package]]
 name = "unicode-ident"
@@ -9037,14 +9078,14 @@ dependencies = [
 
 [[package]]
 name = "utoipa"
-version = "4.1.0"
+version = "4.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ff05e3bac2c9428f57ade702667753ca3f5cf085e2011fe697de5bfd49aa72d"
+checksum = "272ebdfbc99111033031d2f10e018836056e4d2c8e2acda76450ec7974269fa7"
 dependencies = [
  "indexmap 2.1.0",
  "serde",
  "serde_json",
- "utoipa-gen 4.1.0",
+ "utoipa-gen 4.2.0",
 ]
 
 [[package]]
@@ -9056,19 +9097,19 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
 name = "utoipa-gen"
-version = "4.1.0"
+version = "4.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5f0b6f4667edd64be0e820d6631a60433a269710b6ee89ac39525b872b76d61d"
+checksum = "d3c9f4d08338c1bfa70dde39412a040a884c6f318b3d09aaaf3437a1e52027fc"
 dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
@@ -9083,15 +9124,15 @@ dependencies = [
  "rust-embed",
  "serde",
  "serde_json",
- "utoipa 4.1.0",
+ "utoipa 4.2.0",
  "zip",
 ]
 
 [[package]]
 name = "uuid"
-version = "1.6.1"
+version = "1.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5e395fcf16a7a3d8127ec99782007af141946b4795001f876d54fb0d55978560"
+checksum = "f00cc9702ca12d3c81455259621e676d0f7251cec66a21e98fe2e9a37db93b2a"
 dependencies = [
  "getrandom",
  "serde",
@@ -9105,9 +9146,9 @@ checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
 
 [[package]]
 name = "value-bag"
-version = "1.4.2"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a72e1902dde2bd6441347de2b70b7f5d59bf157c6c62f0c44572607a1d55bbe"
+checksum = "7cdbaf5e132e593e9fc1de6a15bbec912395b11fb9719e061cf64f804524c503"
 
 [[package]]
 name = "vcpkg"
@@ -9117,11 +9158,14 @@ checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
 
 [[package]]
 name = "vergen"
-version = "8.2.6"
+version = "8.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1290fd64cc4e7d3c9b07d7f333ce0ce0007253e32870e632624835cc80b83939"
+checksum = "e27d6bdd219887a9eadd19e1c34f32e47fa332301184935c6d9bca26f3cca525"
 dependencies = [
  "anyhow",
+ "cargo_metadata",
+ "cfg-if",
+ "regex",
  "rustversion",
  "time",
 ]
@@ -9171,9 +9215,9 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.89"
+version = "0.2.90"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ed0d4f68a3015cc185aff4db9506a015f4b96f95303897bfa23f846db54064e"
+checksum = "b1223296a201415c7fad14792dbefaace9bd52b62d33453ade1c5b5f07555406"
 dependencies = [
  "cfg-if",
  "wasm-bindgen-macro",
@@ -9181,24 +9225,24 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-backend"
-version = "0.2.89"
+version = "0.2.90"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b56f625e64f3a1084ded111c4d5f477df9f8c92df113852fa5a374dbda78826"
+checksum = "fcdc935b63408d58a32f8cc9738a0bffd8f05cc7c002086c6ef20b7312ad9dcd"
 dependencies = [
  "bumpalo",
  "log",
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
  "wasm-bindgen-shared",
 ]
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.39"
+version = "0.4.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ac36a15a220124ac510204aec1c3e5db8a22ab06fd6706d881dc6149f8ed9a12"
+checksum = "bde2032aeb86bdfaecc8b261eef3cba735cc426c1f3a3416d1e0791be95fc461"
 dependencies = [
  "cfg-if",
  "js-sys",
@@ -9208,9 +9252,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.89"
+version = "0.2.90"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0162dbf37223cd2afce98f3d0785506dcb8d266223983e4b5b525859e6e182b2"
+checksum = "3e4c238561b2d428924c49815533a8b9121c664599558a5d9ec51f8a1740a999"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -9218,28 +9262,28 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.89"
+version = "0.2.90"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f0eb82fcb7930ae6219a7ecfd55b217f5f0893484b7a13022ebb2b2bf20b5283"
+checksum = "bae1abb6806dc1ad9e560ed242107c0f6c84335f1749dd4e8ddb012ebd5e25a7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.89"
+version = "0.2.90"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ab9b36309365056cd639da3134bf87fa8f3d86008abf99e612384a6eecd459f"
+checksum = "4d91413b1c31d7539ba5ef2451af3f0b833a005eb27a631cec32bc0635a8602b"
 
 [[package]]
 name = "wasm-encoder"
-version = "0.38.1"
+version = "0.39.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ad2b51884de9c7f4fe2fd1043fccb8dcad4b1e29558146ee57a144d15779f3f"
+checksum = "111495d6204760238512f57a9af162f45086504da332af210f2f75dd80b34f1d"
 dependencies = [
  "leb128",
 ]
@@ -9318,7 +9362,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f58ddfe801df3886feaf466d883ea37e941bcc6d841b9f644a08c7acabfe7f8"
 dependencies = [
  "anyhow",
- "base64 0.21.5",
+ "base64 0.21.7",
  "bincode 1.3.3",
  "directories-next",
  "file-per-thread-logger",
@@ -9522,9 +9566,9 @@ dependencies = [
 
 [[package]]
 name = "wast"
-version = "69.0.1"
+version = "70.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c1ee37317321afde358e4d7593745942c48d6d17e0e6e943704de9bbee121e7a"
+checksum = "2ee4bc54bbe1c6924160b9f75e374a1d07532e7580eb632c0ee6cdd109bb217e"
 dependencies = [
  "leb128",
  "memchr",
@@ -9534,18 +9578,18 @@ dependencies = [
 
 [[package]]
 name = "wat"
-version = "1.0.82"
+version = "1.0.83"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aeb338ee8dee4d4cd05e6426683f21c5087dc7cfc8903e839ccf48d43332da3c"
+checksum = "9f0dce8cdc288c717cf01e461a1e451a7b8445d53451123536ba576e423a101a"
 dependencies = [
  "wast",
 ]
 
 [[package]]
 name = "web-sys"
-version = "0.3.66"
+version = "0.3.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50c24a44ec86bb68fbecd1b3efed7e85ea5621b39b35ef2766b66cd984f8010f"
+checksum = "58cd2333b6e0be7a39605f0e255892fd7418a682d8da8fe042fe25128794d2ed"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -9586,7 +9630,7 @@ dependencies = [
  "either",
  "home",
  "once_cell",
- "rustix 0.38.28",
+ "rustix 0.38.30",
 ]
 
 [[package]]
@@ -9839,9 +9883,9 @@ checksum = "dff9641d1cd4be8d1a070daf9e3773c5f67e78b4d9d42263020c057706765c04"
 
 [[package]]
 name = "winnow"
-version = "0.5.32"
+version = "0.5.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8434aeec7b290e8da5c3f0d628cb0eac6cabcb31d14bb74f779a08109a5914d6"
+checksum = "b7cf47b659b318dccbd69cc4797a39ae128f533dce7902a1096044d1967b9c16"
 dependencies = [
  "memchr",
 ]
@@ -9929,7 +9973,7 @@ checksum = "9ce1b18ccd8e73a9321186f97e46f9f04b778851177567b1975109d26a08d2a6"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.46",
+ "syn 2.0.48",
 ]
 
 [[package]]
diff --git a/arroyo-api/src/pipelines.rs b/arroyo-api/src/pipelines.rs
index 5e86b1854..67a41800f 100644
--- a/arroyo-api/src/pipelines.rs
+++ b/arroyo-api/src/pipelines.rs
@@ -55,6 +55,7 @@ use crate::types::public::{PipelineType, RestartMode, StopMode};
 use crate::{connection_tables, to_micros};
 use crate::{handle_db_error, AuthData};
 use create_pipeline_req::Config::Sql;
+
 const DEFAULT_CHECKPOINT_INTERVAL: Duration = Duration::from_secs(10);
 
 async fn compile_sql<'e, E>(
diff --git a/arroyo-compiler-service/src/main.rs b/arroyo-compiler-service/src/main.rs
index 292c83327..27564140d 100644
--- a/arroyo-compiler-service/src/main.rs
+++ b/arroyo-compiler-service/src/main.rs
@@ -355,6 +355,13 @@ impl CompilerGrpc for CompileService {
             .await
             .map_err(|e| Status::internal(format!("Writing UDFs failed: {}", e)))?;
 
+        // udfs depend on arroyo-types::UdfContext, so the types crate needs to export it
+        tokio::fs::write(
+            self.build_dir.join("types/src/lib.rs"),
+            "pub use arroyo_types::UdfContext;",
+        )
+        .await?;
+
         let output = Command::new("cargo")
             .current_dir(&udf_build_dir)
             .arg("check")
diff --git a/arroyo-connectors/src/kafka.rs b/arroyo-connectors/src/kafka.rs
index fa705e819..6f9e0c3be 100644
--- a/arroyo-connectors/src/kafka.rs
+++ b/arroyo-connectors/src/kafka.rs
@@ -83,6 +83,7 @@ impl KafkaConnector {
                 TableType::Source {
                     offset: match offset.as_ref().map(|f| f.as_str()) {
                         Some("earliest") => SourceOffset::Earliest,
+                        Some("group") => SourceOffset::Group,
                         None | Some("latest") => SourceOffset::Latest,
                         Some(other) => bail!("invalid value for source.offset '{}'", other),
                     },
diff --git a/arroyo-connectors/src/redis.rs b/arroyo-connectors/src/redis.rs
index 4c951c54c..6db9bd5da 100644
--- a/arroyo-connectors/src/redis.rs
+++ b/arroyo-connectors/src/redis.rs
@@ -1,17 +1,21 @@
+use std::collections::HashMap;
+use std::convert::Infallible;
+
 use anyhow::{anyhow, bail};
-use arroyo_rpc::api_types::connections::{
-    ConnectionProfile, ConnectionSchema, ConnectionType, FieldType, PrimitiveType,
-    TestSourceMessage,
-};
-use arroyo_rpc::OperatorConfig;
+use arroyo_rpc::var_str::VarStr;
 use axum::response::sse::Event;
 use redis::cluster::ClusterClient;
+use redis::{Client, ConnectionInfo, IntoConnectionInfo};
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use std::convert::Infallible;
 use tokio::sync::oneshot::Receiver;
 use typify::import_types;
 
+use arroyo_rpc::api_types::connections::{
+    ConnectionProfile, ConnectionSchema, ConnectionType, FieldType, PrimitiveType,
+    TestSourceMessage,
+};
+use arroyo_rpc::OperatorConfig;
+
 use crate::{pull_opt, pull_option_to_u64, Connection, Connector};
 
 pub struct RedisConnector {}
@@ -20,31 +24,64 @@ const CONFIG_SCHEMA: &str = include_str!("../../connector-schemas/redis/connecti
 const TABLE_SCHEMA: &str = include_str!("../../connector-schemas/redis/table.json");
 const ICON: &str = include_str!("../resources/redis.svg");
 
-import_types!(schema = "../connector-schemas/redis/connection.json",);
+import_types!(
+    schema = "../connector-schemas/redis/connection.json",
+    convert = {
+        {type = "string", format = "var-str"} = VarStr
+    }
+);
 import_types!(schema = "../connector-schemas/redis/table.json");
 
-fn validate_column(
-    schema: &ConnectionSchema,
-    column: String,
-    path: &str,
-) -> anyhow::Result<String> {
-    if schema
-        .fields
-        .iter()
-        .find(|f| {
-            f.field_name == column
-                && f.field_type.r#type == FieldType::Primitive(PrimitiveType::String)
+enum RedisClient {
+    Standard(Client),
+    Clustered(ClusterClient),
+}
+
+impl RedisClient {
+    pub fn new(config: &RedisConfig) -> anyhow::Result<Self> {
+        Ok(match &config.connection {
+            RedisConfigConnection::Address(address) => {
+                let info = from_address(&config, &address.0)?;
+
+                RedisClient::Standard(Client::open(info).map_err(|e| {
+                    anyhow!(
+                        "Failed to construct Redis client for {}: {:?}",
+                        address.0,
+                        e
+                    )
+                })?)
+            }
+            RedisConfigConnection::Addresses(addresses) => {
+                let infos: anyhow::Result<Vec<ConnectionInfo>> = addresses
+                    .iter()
+                    .map(|address| from_address(&config, address))
+                    .collect();
+
+                RedisClient::Clustered(
+                    ClusterClient::new(infos?).map_err(|e| {
+                        anyhow!("Failed to construct Redis Cluster client: {:?}", e)
+                    })?,
+                )
+            }
         })
-        .is_none()
-    {
-        bail!(
-            "invalid value '{}' for {}, must be the name of a TEXT column on the table",
-            column,
-            path
-        );
+    }
+}
+
+fn from_address(config: &RedisConfig, address: &str) -> anyhow::Result<ConnectionInfo> {
+    let mut info: ConnectionInfo = address
+        .to_string()
+        .into_connection_info()
+        .map_err(|e| anyhow!("invalid redis address: {:?}", e))?;
+
+    if let Some(username) = &config.username {
+        info.redis.username = Some(username.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
+    }
+
+    if let Some(password) = &config.password {
+        info.redis.password = Some(password.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
     }
 
-    Ok(column)
+    Ok(info)
 }
 
 async fn test_inner(
@@ -57,21 +94,14 @@ async fn test_inner(
         .await
         .unwrap();
 
-    match c.connection {
-        RedisConfigConnection::Address(address) => {
-            let client = redis::Client::open(address.0.clone()).map_err(|e| {
-                anyhow!(
-                    "Failed to construct Redis client for {}: {:?}",
-                    address.0,
-                    e
-                )
-            })?;
+    let client = RedisClient::new(&c)?;
 
+    match &client {
+        RedisClient::Standard(client) => {
             let mut connection = client
                 .get_async_connection()
                 .await
-                .map_err(|e| anyhow!("Failed to connect to Redis at {}: {:?}", address.0, e))?;
-
+                .map_err(|e| anyhow!("Failed to connect to to Redis Cluster: {:?}", e))?;
             tx.send(Ok(Event::default()
                 .json_data(TestSourceMessage::info(
                     "Connected successfully, sending PING",
@@ -85,15 +115,11 @@ async fn test_inner(
                 .await
                 .map_err(|e| anyhow!("Received error sending PING command: {:?}", e))?;
         }
-        RedisConfigConnection::Addresses(addresses) => {
-            let client = ClusterClient::new(addresses.into_iter().map(|a| a.0).collect())
-                .map_err(|e| anyhow!("Failed to construct Redis Cluster client: {:?}", e))?;
-
+        RedisClient::Clustered(client) => {
             let mut connection = client
                 .get_async_connection()
                 .await
                 .map_err(|e| anyhow!("Failed to connect to to Redis Cluster: {:?}", e))?;
-
             tx.send(Ok(Event::default()
                 .json_data(TestSourceMessage::info(
                     "Connected successfully, sending PING",
@@ -107,7 +133,7 @@ async fn test_inner(
                 .await
                 .map_err(|e| anyhow!("Received error sending PING command: {:?}", e))?;
         }
-    };
+    }
 
     Ok("Received PING response successfully".to_string())
 }
@@ -154,7 +180,7 @@ impl Connector for RedisConnector {
         let (tx, rx) = tokio::sync::oneshot::channel();
 
         tokio::spawn(async move {
-            let (itx, _) = tokio::sync::mpsc::channel(8);
+            let (itx, _rx) = tokio::sync::mpsc::channel(8);
             let message = match test_inner(profile, itx).await {
                 Ok(_) => TestSourceMessage::done("Successfully connected to Redis"),
                 Err(e) => TestSourceMessage::fail(format!("Failed to connect to Redis: {:?}", e)),
@@ -195,18 +221,16 @@ impl Connector for RedisConnector {
     ) -> anyhow::Result<Connection> {
         let connection_config = match profile {
             Some(connection_profile) => {
-                let config: RedisConfigConnection =
-                    serde_json::from_value(connection_profile.config.clone())
-                        .map_err(|e| anyhow!("Failed to parse connection config: {:?}", e))?;
-                RedisConfig { connection: config }
+                serde_json::from_value(connection_profile.config.clone())
+                    .map_err(|e| anyhow!("Failed to parse connection config: {:?}", e))?
             }
             None => {
                 let address = options.remove("address");
 
                 let cluster_addresses = options.remove("cluster.addresses");
-                let connection_config = match (address, cluster_addresses) {
+                let connection = match (address, cluster_addresses) {
                     (Some(address), None) => {
-                        RedisConfigConnection::Address(RedisConfigConnectionAddress(address))
+                        RedisConfigConnection::Address(Address(address.to_string()))
                     }
                     (None, Some(cluster_addresses)) => RedisConfigConnection::Addresses(
                         cluster_addresses
@@ -221,8 +245,14 @@ impl Connector for RedisConnector {
                         bail!("one of `address` or `cluster.addresses` must be set");
                     }
                 };
+
+                let username = options.remove("username").map(VarStr::new);
+                let password = options.remove("password").map(VarStr::new);
+
                 RedisConfig {
-                    connection: connection_config,
+                    connection,
+                    username,
+                    password,
                 }
             }
         };
@@ -339,51 +369,7 @@ impl Connector for RedisConnector {
             .map(|t| t.to_owned())
             .ok_or_else(|| anyhow!("'format' must be set for Redis connection"))?;
 
-        match &config.connection {
-            RedisConfigConnection::Address(address) => {
-                let _ = redis::Client::open(address.0.clone()).map_err(|e| {
-                    anyhow!(
-                        "Failed to construct Redis client for {}: {:?}",
-                        address.0,
-                        e
-                    )
-                })?;
-            }
-            RedisConfigConnection::Addresses(addresses) => {
-                let _ = ClusterClient::new(addresses.into_iter().map(|a| a.0.clone()).collect())
-                    .map_err(|e| anyhow!("Failed to construct Redis Cluster client: {:?}", e))?;
-            }
-        }
-
-        match &table.connector_type {
-            TableType::Target(Target::StringTable { key_column, .. }) => {
-                if let Some(key_column) = key_column {
-                    validate_column(&schema, key_column.clone(), "connector_type.key_column")?;
-                }
-            }
-            TableType::Target(Target::ListTable {
-                list_key_column, ..
-            }) => {
-                if let Some(n) = list_key_column {
-                    validate_column(&schema, n.clone(), "connector_type.list_key_column")?;
-                }
-            }
-            TableType::Target(Target::HashTable {
-                hash_key_column,
-                hash_field_column,
-                ..
-            }) => {
-                if let Some(n) = hash_key_column {
-                    validate_column(&schema, n.clone(), "connector_type.hash_key_column")?;
-                }
-
-                validate_column(
-                    &schema,
-                    hash_field_column.clone(),
-                    "connector_type.hash_field_column",
-                )?;
-            }
-        };
+        let _ = RedisClient::new(&config)?;
 
         let config = OperatorConfig {
             connection: serde_json::to_value(config).unwrap(),
diff --git a/arroyo-connectors/src/webhook.rs b/arroyo-connectors/src/webhook.rs
index 876db528e..0ee25ef62 100644
--- a/arroyo-connectors/src/webhook.rs
+++ b/arroyo-connectors/src/webhook.rs
@@ -29,7 +29,7 @@ pub struct WebhookConnector {}
 impl WebhookConnector {
     fn construct_test_request(client: &Client, config: &WebhookTable) -> anyhow::Result<Request> {
         let req = client
-            .post(&config.endpoint)
+            .post(&config.endpoint.sub_env_vars()?)
             // TODO: use the schema to construct a correctly-formatted message
             .body(
                 serde_json::to_string(&json! {{
@@ -53,7 +53,7 @@ impl WebhookConnector {
             .map(|s| s.sub_env_vars())
             .transpose()?;
 
-        let client = construct_http_client(&config.endpoint, headers)?;
+        let client = construct_http_client(&config.endpoint.sub_env_vars()?, headers)?;
         let req = Self::construct_test_request(&client, config)?;
 
         tx.send(Ok(Event::default()
@@ -141,7 +141,7 @@ impl Connector for WebhookConnector {
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
     ) -> anyhow::Result<crate::Connection> {
-        let description = format!("WebhookSource<{}>", table.endpoint);
+        let description = format!("WebhookSink<{}>", table.endpoint.sub_env_vars()?);
 
         let schema = schema
             .map(|s| s.to_owned())
@@ -184,10 +184,13 @@ impl Connector for WebhookConnector {
 
         let headers = options.remove("headers").map(|s| VarStr::new(s));
 
-        let table = WebhookTable { endpoint, headers };
+        let table = WebhookTable {
+            endpoint: VarStr::new(endpoint),
+            headers,
+        };
 
         let client = construct_http_client(
-            &table.endpoint,
+            &table.endpoint.sub_env_vars()?,
             table
                 .headers
                 .as_ref()
diff --git a/arroyo-console/package.json b/arroyo-console/package.json
index b3d71fe30..b18888eb6 100644
--- a/arroyo-console/package.json
+++ b/arroyo-console/package.json
@@ -38,7 +38,7 @@
     "dagre": "^0.8.5",
     "downshift": "^8.2.3",
     "formik": "^2.4.2",
-    "framer-motion": "^7.10.3",
+    "framer-motion": "^10.18.0",
     "lodash": "^4.17.21",
     "metrics-graphics": "3.0.0-beta1",
     "monaco-editor": "^0.34.1",
diff --git a/arroyo-console/pnpm-lock.yaml b/arroyo-console/pnpm-lock.yaml
index 123227bd1..f2df40a94 100644
--- a/arroyo-console/pnpm-lock.yaml
+++ b/arroyo-console/pnpm-lock.yaml
@@ -1,9 +1,5 @@
 lockfileVersion: '6.0'
 
-settings:
-  autoInstallPeers: true
-  excludeLinksFromLockfile: false
-
 dependencies:
   '@babel/core':
     specifier: ^7.22.5
@@ -19,7 +15,7 @@ dependencies:
     version: 0.0.57(@chakra-ui/react@2.8.0)(@chakra-ui/styled-system@2.9.1)
   '@chakra-ui/react':
     specifier: ^2.8.0
-    version: 2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0)
+    version: 2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0)
   '@chakra-ui/stepper':
     specifier: ^2.2.0
     version: 2.2.0(@chakra-ui/system@2.6.2)(react@18.2.0)
@@ -40,7 +36,7 @@ dependencies:
     version: 4.5.1(monaco-editor@0.34.1)(react-dom@18.2.0)(react@18.2.0)
   '@rjsf/chakra-ui':
     specifier: ^5.8.1
-    version: 5.8.1(@chakra-ui/icons@2.0.19)(@chakra-ui/react@2.8.0)(@chakra-ui/system@2.6.2)(@rjsf/core@5.8.1)(@rjsf/utils@5.8.1)(@types/react@18.2.12)(chakra-react-select@4.7.6)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0)
+    version: 5.8.1(@chakra-ui/icons@2.0.19)(@chakra-ui/react@2.8.0)(@chakra-ui/system@2.6.2)(@rjsf/core@5.8.1)(@rjsf/utils@5.8.1)(@types/react@18.2.12)(chakra-react-select@4.7.6)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0)
   '@rjsf/core':
     specifier: ^5.8.1
     version: 5.8.1(@rjsf/utils@5.8.1)(react@18.2.0)
@@ -84,8 +80,8 @@ dependencies:
     specifier: ^2.4.2
     version: 2.4.2(react@18.2.0)
   framer-motion:
-    specifier: ^7.10.3
-    version: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+    specifier: ^10.18.0
+    version: 10.18.0(react-dom@18.2.0)(react@18.2.0)
   lodash:
     specifier: ^4.17.21
     version: 4.17.21
@@ -485,7 +481,7 @@ packages:
       '@babel/helper-validator-identifier': 7.22.20
       to-fast-properties: 2.0.0
 
-  /@chakra-ui/accordion@2.3.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0):
+  /@chakra-ui/accordion@2.3.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0):
     resolution: {integrity: sha512-A4TkRw3Jnt+Fam6dSSJ62rskdrvjF3JGctYcfXlojfFIpHPuIw4pDwfZgNAxlaxWkcj0e7JJKlQ88dnZW+QfFg==}
     peerDependencies:
       '@chakra-ui/system': '>=2.0.0'
@@ -499,8 +495,8 @@ packages:
       '@chakra-ui/react-use-merge-refs': 2.1.0(react@18.2.0)
       '@chakra-ui/shared-utils': 2.0.5
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
-      '@chakra-ui/transition': 2.1.0(framer-motion@7.10.3)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/transition': 2.1.0(framer-motion@10.18.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
     dev: false
 
@@ -919,7 +915,7 @@ packages:
       react: 18.2.0
     dev: false
 
-  /@chakra-ui/menu@2.2.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0):
+  /@chakra-ui/menu@2.2.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0):
     resolution: {integrity: sha512-l7HQjriW4JGeCyxDdguAzekwwB+kHGDLxACi0DJNp37sil51SRaN1S1OrneISbOHVpHuQB+KVNgU0rqhoglVew==}
     peerDependencies:
       '@chakra-ui/system': '>=2.0.0'
@@ -941,12 +937,12 @@ packages:
       '@chakra-ui/react-use-update-effect': 2.1.0(react@18.2.0)
       '@chakra-ui/shared-utils': 2.0.5
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
-      '@chakra-ui/transition': 2.1.0(framer-motion@7.10.3)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/transition': 2.1.0(framer-motion@10.18.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
     dev: false
 
-  /@chakra-ui/menu@2.2.1(@chakra-ui/system@2.6.2)(framer-motion@7.10.3)(react@18.2.0):
+  /@chakra-ui/menu@2.2.1(@chakra-ui/system@2.6.2)(framer-motion@10.18.0)(react@18.2.0):
     resolution: {integrity: sha512-lJS7XEObzJxsOwWQh7yfG4H8FzFPRP5hVPN/CL+JzytEINCSBvsCDHrYPQGp7jzpCi8vnTqQQGQe0f8dwnXd2g==}
     peerDependencies:
       '@chakra-ui/system': '>=2.0.0'
@@ -968,12 +964,12 @@ packages:
       '@chakra-ui/react-use-update-effect': 2.1.0(react@18.2.0)
       '@chakra-ui/shared-utils': 2.0.5
       '@chakra-ui/system': 2.6.2(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
-      '@chakra-ui/transition': 2.1.0(framer-motion@7.10.3)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/transition': 2.1.0(framer-motion@10.18.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
     dev: false
 
-  /@chakra-ui/modal@2.3.0(@chakra-ui/system@2.6.0)(@types/react@18.2.12)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0):
+  /@chakra-ui/modal@2.3.0(@chakra-ui/system@2.6.0)(@types/react@18.2.12)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0):
     resolution: {integrity: sha512-S1sITrIeLSf21LJ0Vz8xZhj5fWEud5z5Dl2dmvOEv1ezypgOrCCBdOEnnqCkoEKZDbKvzZWZXWR5791ikLP6+g==}
     peerDependencies:
       '@chakra-ui/system': '>=2.0.0'
@@ -989,9 +985,9 @@ packages:
       '@chakra-ui/react-use-merge-refs': 2.1.0(react@18.2.0)
       '@chakra-ui/shared-utils': 2.0.5
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
-      '@chakra-ui/transition': 2.1.0(framer-motion@7.10.3)(react@18.2.0)
+      '@chakra-ui/transition': 2.1.0(framer-motion@10.18.0)(react@18.2.0)
       aria-hidden: 1.2.3
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
       react-dom: 18.2.0(react@18.2.0)
       react-remove-scroll: 2.5.6(@types/react@18.2.12)(react@18.2.0)
@@ -1045,7 +1041,7 @@ packages:
       react: 18.2.0
     dev: false
 
-  /@chakra-ui/popover@2.2.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0):
+  /@chakra-ui/popover@2.2.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0):
     resolution: {integrity: sha512-cTqXdgkU0vgK82AR1nWcC2MJYhEL/y6uTeprvO2+j4o2D0yPrzVMuIZZRl0abrQwiravQyVGEMgA5y0ZLYwbiQ==}
     peerDependencies:
       '@chakra-ui/system': '>=2.0.0'
@@ -1064,7 +1060,7 @@ packages:
       '@chakra-ui/react-use-merge-refs': 2.1.0(react@18.2.0)
       '@chakra-ui/shared-utils': 2.0.5
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
     dev: false
 
@@ -1096,7 +1092,7 @@ packages:
     peerDependencies:
       '@chakra-ui/react': 1.8.x || 2.x
     dependencies:
-      '@chakra-ui/react': 2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/react': 2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0)
       '@chakra-ui/theme-tools': 2.0.12(@chakra-ui/styled-system@2.9.1)
     transitivePeerDependencies:
       - '@chakra-ui/styled-system'
@@ -1351,7 +1347,7 @@ packages:
       react: 18.2.0
     dev: false
 
-  /@chakra-ui/react@2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0):
+  /@chakra-ui/react@2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0):
     resolution: {integrity: sha512-tV82DaqE4fMbLIWq58BYh4Ol3gAlNEn+qYOzx8bPrZudboEDnboq8aVfSBwWOY++MLWz2Nn7CkT69YRm91e5sg==}
     peerDependencies:
       '@emotion/react': ^11.0.0
@@ -1360,7 +1356,7 @@ packages:
       react: '>=18'
       react-dom: '>=18'
     dependencies:
-      '@chakra-ui/accordion': 2.3.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0)
+      '@chakra-ui/accordion': 2.3.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0)
       '@chakra-ui/alert': 2.2.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/avatar': 2.3.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/breadcrumb': 2.2.0(@chakra-ui/system@2.6.0)(react@18.2.0)
@@ -1381,11 +1377,11 @@ packages:
       '@chakra-ui/layout': 2.3.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/live-region': 2.1.0(react@18.2.0)
       '@chakra-ui/media-query': 3.3.0(@chakra-ui/system@2.6.0)(react@18.2.0)
-      '@chakra-ui/menu': 2.2.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0)
-      '@chakra-ui/modal': 2.3.0(@chakra-ui/system@2.6.0)(@types/react@18.2.12)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/menu': 2.2.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0)
+      '@chakra-ui/modal': 2.3.0(@chakra-ui/system@2.6.0)(@types/react@18.2.12)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0)
       '@chakra-ui/number-input': 2.1.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/pin-input': 2.1.0(@chakra-ui/system@2.6.0)(react@18.2.0)
-      '@chakra-ui/popover': 2.2.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0)
+      '@chakra-ui/popover': 2.2.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0)
       '@chakra-ui/popper': 3.1.0(react@18.2.0)
       '@chakra-ui/portal': 2.1.0(react-dom@18.2.0)(react@18.2.0)
       '@chakra-ui/progress': 2.2.0(@chakra-ui/system@2.6.0)(react@18.2.0)
@@ -1400,7 +1396,7 @@ packages:
       '@chakra-ui/stat': 2.1.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/stepper': 2.3.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/styled-system': 2.9.1
-      '@chakra-ui/switch': 2.1.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0)
+      '@chakra-ui/switch': 2.1.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0)
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
       '@chakra-ui/table': 2.1.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/tabs': 2.2.0(@chakra-ui/system@2.6.0)(react@18.2.0)
@@ -1408,14 +1404,14 @@ packages:
       '@chakra-ui/textarea': 2.1.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/theme': 3.2.0(@chakra-ui/styled-system@2.9.1)
       '@chakra-ui/theme-utils': 2.0.19
-      '@chakra-ui/toast': 7.0.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0)
-      '@chakra-ui/tooltip': 2.3.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0)
-      '@chakra-ui/transition': 2.1.0(framer-motion@7.10.3)(react@18.2.0)
+      '@chakra-ui/toast': 7.0.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/tooltip': 2.3.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/transition': 2.1.0(framer-motion@10.18.0)(react@18.2.0)
       '@chakra-ui/utils': 2.0.15
       '@chakra-ui/visually-hidden': 2.1.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@emotion/react': 11.11.1(@types/react@18.2.12)(react@18.2.0)
       '@emotion/styled': 11.11.0(@emotion/react@11.11.1)(@types/react@18.2.12)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
       react-dom: 18.2.0(react@18.2.0)
     transitivePeerDependencies:
@@ -1558,7 +1554,7 @@ packages:
       lodash.mergewith: 4.6.2
     dev: false
 
-  /@chakra-ui/switch@2.1.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react@18.2.0):
+  /@chakra-ui/switch@2.1.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react@18.2.0):
     resolution: {integrity: sha512-uWHOaIDQdGh+mszxeppj5aYVepbkSK445KZlJJkfr9Bnr6sythTwM63HSufnVDiTEE4uRqegv9jEjZK2JKA+9A==}
     peerDependencies:
       '@chakra-ui/system': '>=2.0.0'
@@ -1568,7 +1564,7 @@ packages:
       '@chakra-ui/checkbox': 2.3.0(@chakra-ui/system@2.6.0)(react@18.2.0)
       '@chakra-ui/shared-utils': 2.0.5
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
     dev: false
 
@@ -1737,7 +1733,7 @@ packages:
       '@chakra-ui/theme-tools': 2.1.2(@chakra-ui/styled-system@2.9.2)
     dev: false
 
-  /@chakra-ui/toast@7.0.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0):
+  /@chakra-ui/toast@7.0.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0):
     resolution: {integrity: sha512-XQgSnn4DYRgfOBzBvh8GI/AZ7SfrO8wlVSmChfp92Nfmqm7tRDUT9x8ws/iNKAvMRHkhl7fmRjJ39ipeXYrMvA==}
     peerDependencies:
       '@chakra-ui/system': 2.6.0
@@ -1755,12 +1751,12 @@ packages:
       '@chakra-ui/styled-system': 2.9.1
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
       '@chakra-ui/theme': 3.2.0(@chakra-ui/styled-system@2.9.1)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
       react-dom: 18.2.0(react@18.2.0)
     dev: false
 
-  /@chakra-ui/tooltip@2.3.0(@chakra-ui/system@2.6.0)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0):
+  /@chakra-ui/tooltip@2.3.0(@chakra-ui/system@2.6.0)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0):
     resolution: {integrity: sha512-2s23f93YIij1qEDwIK//KtEu4LLYOslhR1cUhDBk/WUzyFR3Ez0Ee+HlqlGEGfGe9x77E6/UXPnSAKKdF/cpsg==}
     peerDependencies:
       '@chakra-ui/system': '>=2.0.0'
@@ -1777,19 +1773,19 @@ packages:
       '@chakra-ui/react-use-merge-refs': 2.1.0(react@18.2.0)
       '@chakra-ui/shared-utils': 2.0.5
       '@chakra-ui/system': 2.6.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
       react-dom: 18.2.0(react@18.2.0)
     dev: false
 
-  /@chakra-ui/transition@2.1.0(framer-motion@7.10.3)(react@18.2.0):
+  /@chakra-ui/transition@2.1.0(framer-motion@10.18.0)(react@18.2.0):
     resolution: {integrity: sha512-orkT6T/Dt+/+kVwJNy7zwJ+U2xAZ3EU7M3XCs45RBvUnZDr/u9vdmaM/3D/rOpmQJWgQBwKPJleUXrYWUagEDQ==}
     peerDependencies:
       framer-motion: '>=4.0.0'
       react: '>=18'
     dependencies:
       '@chakra-ui/shared-utils': 2.0.5
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
     dev: false
 
@@ -2290,53 +2286,6 @@ packages:
       react-dom: 18.2.0(react@18.2.0)
     dev: false
 
-  /@motionone/animation@10.16.3:
-    resolution: {integrity: sha512-QUGWpLbMFLhyqKlngjZhjtxM8IqiJQjLK0DF+XOF6od9nhSvlaeEpOY/UMCRVcZn/9Tr2rZO22EkuCIjYdI74g==}
-    dependencies:
-      '@motionone/easing': 10.16.3
-      '@motionone/types': 10.16.3
-      '@motionone/utils': 10.16.3
-      tslib: 2.5.3
-    dev: false
-
-  /@motionone/dom@10.16.4:
-    resolution: {integrity: sha512-HPHlVo/030qpRj9R8fgY50KTN4Ko30moWRTA3L3imrsRBmob93cTYmodln49HYFbQm01lFF7X523OkKY0DX6UA==}
-    dependencies:
-      '@motionone/animation': 10.16.3
-      '@motionone/generators': 10.16.4
-      '@motionone/types': 10.16.3
-      '@motionone/utils': 10.16.3
-      hey-listen: 1.0.8
-      tslib: 2.5.3
-    dev: false
-
-  /@motionone/easing@10.16.3:
-    resolution: {integrity: sha512-HWTMZbTmZojzwEuKT/xCdvoMPXjYSyQvuVM6jmM0yoGU6BWzsmYMeB4bn38UFf618fJCNtP9XeC/zxtKWfbr0w==}
-    dependencies:
-      '@motionone/utils': 10.16.3
-      tslib: 2.5.3
-    dev: false
-
-  /@motionone/generators@10.16.4:
-    resolution: {integrity: sha512-geFZ3w0Rm0ZXXpctWsSf3REGywmLLujEjxPYpBR0j+ymYwof0xbV6S5kGqqsDKgyWKVWpUInqQYvQfL6fRbXeg==}
-    dependencies:
-      '@motionone/types': 10.16.3
-      '@motionone/utils': 10.16.3
-      tslib: 2.5.3
-    dev: false
-
-  /@motionone/types@10.16.3:
-    resolution: {integrity: sha512-W4jkEGFifDq73DlaZs3HUfamV2t1wM35zN/zX7Q79LfZ2sc6C0R1baUHZmqc/K5F3vSw3PavgQ6HyHLd/MXcWg==}
-    dev: false
-
-  /@motionone/utils@10.16.3:
-    resolution: {integrity: sha512-WNWDksJIxQkaI9p9Z9z0+K27xdqISGNFy1SsWVGaiedTHq0iaT6iZujby8fT/ZnZxj1EOaxJtSfUPCFNU5CRoA==}
-    dependencies:
-      '@motionone/types': 10.16.3
-      hey-listen: 1.0.8
-      tslib: 2.5.3
-    dev: false
-
   /@nodelib/fs.scandir@2.1.5:
     resolution: {integrity: sha512-vq24Bq3ym5HEQm2NKCr3yXDwjc7vTsEThRDnkp2DK9p1uqLR+DHurm/NOTo0KG7HYHU7eppKZj3MyqYuMBf62g==}
     engines: {node: '>= 8'}
@@ -2466,7 +2415,7 @@ packages:
     engines: {node: '>=14'}
     dev: false
 
-  /@rjsf/chakra-ui@5.8.1(@chakra-ui/icons@2.0.19)(@chakra-ui/react@2.8.0)(@chakra-ui/system@2.6.2)(@rjsf/core@5.8.1)(@rjsf/utils@5.8.1)(@types/react@18.2.12)(chakra-react-select@4.7.6)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0):
+  /@rjsf/chakra-ui@5.8.1(@chakra-ui/icons@2.0.19)(@chakra-ui/react@2.8.0)(@chakra-ui/system@2.6.2)(@rjsf/core@5.8.1)(@rjsf/utils@5.8.1)(@types/react@18.2.12)(chakra-react-select@4.7.6)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0):
     resolution: {integrity: sha512-gk0PBaasaayhQXWX5z3M7dnFt5tZi6OEpZ3tq0pr5cpBQNsJ5lEsTJU8sFO8ZaBPBRtf2+aTjOs/bgNGX1rYNA==}
     engines: {node: '>=14'}
     peerDependencies:
@@ -2480,12 +2429,12 @@ packages:
       react: ^16.14.0 || >=17
     dependencies:
       '@chakra-ui/icons': 2.0.19(@chakra-ui/system@2.6.2)(react@18.2.0)
-      '@chakra-ui/react': 2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@7.10.3)(react-dom@18.2.0)(react@18.2.0)
+      '@chakra-ui/react': 2.8.0(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(@types/react@18.2.12)(framer-motion@10.18.0)(react-dom@18.2.0)(react@18.2.0)
       '@chakra-ui/system': 2.6.2(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
       '@rjsf/core': 5.8.1(@rjsf/utils@5.8.1)(react@18.2.0)
       '@rjsf/utils': 5.8.1(react@18.2.0)
       chakra-react-select: 4.7.6(@chakra-ui/form-control@2.2.0)(@chakra-ui/icon@3.2.0)(@chakra-ui/layout@2.3.1)(@chakra-ui/media-query@3.3.0)(@chakra-ui/menu@2.2.1)(@chakra-ui/spinner@2.1.0)(@chakra-ui/system@2.6.2)(@emotion/react@11.11.1)(@types/react@18.2.12)(react-dom@18.2.0)(react@18.2.0)
-      framer-motion: 7.10.3(react-dom@18.2.0)(react@18.2.0)
+      framer-motion: 10.18.0(react-dom@18.2.0)(react@18.2.0)
       react: 18.2.0
       react-select: 5.7.3(@types/react@18.2.12)(react-dom@18.2.0)(react@18.2.0)
     transitivePeerDependencies:
@@ -3096,7 +3045,7 @@ packages:
     resolution: {integrity: sha512-xcLxITLe2HYa1cnYnwCjkOO1PqUHQpozB8x9AR0OgWN2woOBi5kSDVxKfd0b7sb1hw5qFeJhXm9H1nu3xSfLeQ==}
     engines: {node: '>=10'}
     dependencies:
-      tslib: 2.5.3
+      tslib: 2.6.2
     dev: false
 
   /array-buffer-byte-length@1.0.0:
@@ -3241,7 +3190,7 @@ packages:
       '@chakra-ui/icon': 3.2.0(@chakra-ui/system@2.6.2)(react@18.2.0)
       '@chakra-ui/layout': 2.3.1(@chakra-ui/system@2.6.2)(react@18.2.0)
       '@chakra-ui/media-query': 3.3.0(@chakra-ui/system@2.6.2)(react@18.2.0)
-      '@chakra-ui/menu': 2.2.1(@chakra-ui/system@2.6.2)(framer-motion@7.10.3)(react@18.2.0)
+      '@chakra-ui/menu': 2.2.1(@chakra-ui/system@2.6.2)(framer-motion@10.18.0)(react@18.2.0)
       '@chakra-ui/spinner': 2.1.0(@chakra-ui/system@2.6.2)(react@18.2.0)
       '@chakra-ui/system': 2.6.2(@emotion/react@11.11.1)(@emotion/styled@11.11.0)(react@18.2.0)
       '@emotion/react': 11.11.1(@types/react@18.2.12)(react@18.2.0)
@@ -4247,7 +4196,7 @@ packages:
     resolution: {integrity: sha512-KSuV3ur4gf2KqMNoZx3nXNVhqCkn42GuTYCX4tXPEwf0MjpFQmNMiN6m7dXaUXgIoivL6/65agoUMg4RLS0Vbg==}
     engines: {node: '>=10'}
     dependencies:
-      tslib: 2.5.3
+      tslib: 2.6.2
     dev: false
 
   /for-each@0.3.3:
@@ -4276,17 +4225,20 @@ packages:
       tslib: 2.5.3
     dev: false
 
-  /framer-motion@7.10.3(react-dom@18.2.0)(react@18.2.0):
-    resolution: {integrity: sha512-k2ccYeZNSpPg//HTaqrU+4pRq9f9ZpaaN7rr0+Rx5zA4wZLbk547wtDzge2db1sB+1mnJ6r59P4xb+aEIi/W+w==}
+  /framer-motion@10.18.0(react-dom@18.2.0)(react@18.2.0):
+    resolution: {integrity: sha512-oGlDh1Q1XqYPksuTD/usb0I70hq95OUzmL9+6Zd+Hs4XV0oaISBa/UUMSjYiq6m8EUF32132mOJ8xVZS+I0S6w==}
     peerDependencies:
       react: ^18.0.0
       react-dom: ^18.0.0
+    peerDependenciesMeta:
+      react:
+        optional: true
+      react-dom:
+        optional: true
     dependencies:
-      '@motionone/dom': 10.16.4
-      hey-listen: 1.0.8
       react: 18.2.0
       react-dom: 18.2.0(react@18.2.0)
-      tslib: 2.4.0
+      tslib: 2.6.2
     optionalDependencies:
       '@emotion/is-prop-valid': 0.8.8
     dev: false
@@ -4502,10 +4454,6 @@ packages:
       space-separated-tokens: 1.1.5
     dev: false
 
-  /hey-listen@1.0.8:
-    resolution: {integrity: sha512-COpmrF2NOg4TBWUJ5UVyaCU2A88wEMkUPK4hNqyCkqHbxT92BbvfjoSozkAIIm6XhicGlJHhFdullInrdhwU8Q==}
-    dev: false
-
   /highlight.js@10.7.3:
     resolution: {integrity: sha512-tzcUFauisWKNHaRkN4Wjl/ZA07gENAjFl3J/c480dprkGTg5EQstgaNFqBfUqCq54kZRIEcreTsAgF/m2quD7A==}
     dev: false
@@ -5489,7 +5437,7 @@ packages:
       '@types/react': 18.2.12
       react: 18.2.0
       react-style-singleton: 2.2.1(@types/react@18.2.12)(react@18.2.0)
-      tslib: 2.5.3
+      tslib: 2.6.2
     dev: false
 
   /react-remove-scroll@2.5.6(@types/react@18.2.12)(react@18.2.0):
@@ -5506,7 +5454,7 @@ packages:
       react: 18.2.0
       react-remove-scroll-bar: 2.3.4(@types/react@18.2.12)(react@18.2.0)
       react-style-singleton: 2.2.1(@types/react@18.2.12)(react@18.2.0)
-      tslib: 2.5.3
+      tslib: 2.6.2
       use-callback-ref: 1.3.0(@types/react@18.2.12)(react@18.2.0)
       use-sidecar: 1.1.2(@types/react@18.2.12)(react@18.2.0)
     dev: false
@@ -5610,7 +5558,7 @@ packages:
       get-nonce: 1.0.1
       invariant: 2.2.4
       react: 18.2.0
-      tslib: 2.5.3
+      tslib: 2.6.2
     dev: false
 
   /react-syntax-highlighter@15.5.0(react@18.2.0):
@@ -6149,7 +6097,7 @@ packages:
     dependencies:
       '@types/react': 18.2.12
       react: 18.2.0
-      tslib: 2.5.3
+      tslib: 2.6.2
     dev: false
 
   /use-isomorphic-layout-effect@1.1.2(@types/react@18.2.12)(react@18.2.0):
@@ -6186,7 +6134,7 @@ packages:
       '@types/react': 18.2.12
       detect-node-es: 1.1.0
       react: 18.2.0
-      tslib: 2.5.3
+      tslib: 2.6.2
     dev: false
 
   /use-sync-external-store@1.2.0(react@18.2.0):
@@ -6351,3 +6299,7 @@ packages:
       react: 18.2.0
       use-sync-external-store: 1.2.0(react@18.2.0)
     dev: false
+
+settings:
+  autoInstallPeers: true
+  excludeLinksFromLockfile: false
diff --git a/arroyo-console/src/lib/example_queries.ts b/arroyo-console/src/lib/example_queries.ts
index 7c0b4297b..b2f713590 100644
--- a/arroyo-console/src/lib/example_queries.ts
+++ b/arroyo-console/src/lib/example_queries.ts
@@ -20,7 +20,7 @@ export const exampleQueries: ExampleQuery[] = [
       ') WITH (\n' +
       "    connector = 'sse',\n" +
       "    format = 'raw_string',\n" +
-      "    endpoint = 'https://mstdn.party/api/v1/streaming/public',\n" +
+      "    endpoint = 'http://mastodon.arroyo.dev/api/v1/streaming/public',\n" +
       "    events = 'update'\n" +
       ');\n' +
       '\n' +
diff --git a/arroyo-controller/src/compiler.rs b/arroyo-controller/src/compiler.rs
index fea839455..7da70acc4 100644
--- a/arroyo-controller/src/compiler.rs
+++ b/arroyo-controller/src/compiler.rs
@@ -91,6 +91,7 @@ impl ProgramCompiler {
 
         quote! {
             use std::time::SystemTime;
+            pub use arroyo_types::UdfContext;
 
             #(#structs )*
         }
@@ -106,6 +107,7 @@ impl ProgramCompiler {
             use arroyo_worker::operators::sinks;
             use arroyo_worker::operators::joins::*;
             use arroyo_worker::operators::windows::*;
+            use arroyo_worker::operators::async_map::*;
             use arroyo_worker::engine::{Program, SubtaskNode};
             use arroyo_worker::{LogicalEdge, LogicalNode};
             use arroyo_types::RecordBatchBuilder;
diff --git a/arroyo-controller/src/job_controller/mod.rs b/arroyo-controller/src/job_controller/mod.rs
index 6c4a5f55d..c8a09cfaa 100644
--- a/arroyo-controller/src/job_controller/mod.rs
+++ b/arroyo-controller/src/job_controller/mod.rs
@@ -762,11 +762,7 @@ impl JobController {
         let cur_epoch = self.model.epoch;
 
         tokio::spawn(async move {
-            let checkpoint = StateBackend::load_checkpoint_metadata(&job_id, cur_epoch)
-                .await
-                .ok_or_else(|| {
-                    anyhow::anyhow!("Couldn't find checkpoint for job during cleaning")
-                })?;
+            let checkpoint = StateBackend::load_checkpoint_metadata(&job_id, cur_epoch).await?;
 
             let c = pool.get().await?;
             controller_queries::mark_compacting()
diff --git a/arroyo-controller/src/lib.rs b/arroyo-controller/src/lib.rs
index 2d4d6aec0..cba71859e 100644
--- a/arroyo-controller/src/lib.rs
+++ b/arroyo-controller/src/lib.rs
@@ -687,6 +687,8 @@ impl ControllerServer {
 }
 
 fn cargo_toml(name: &str, dependencies: &str) -> String {
+    let types_dep = "types = { path = \"../../types\" }";
+
     format!(
         r#"
 [package]
@@ -694,9 +696,10 @@ name = "{}"
 version = "1.0.0"
 edition = "2021"
 
+{}
 {}
         "#,
-        name, dependencies
+        name, dependencies, types_dep
     )
 }
 
diff --git a/arroyo-controller/src/states/mod.rs b/arroyo-controller/src/states/mod.rs
index 7fdad266e..dd80aaa49 100644
--- a/arroyo-controller/src/states/mod.rs
+++ b/arroyo-controller/src/states/mod.rs
@@ -668,6 +668,8 @@ impl StateMachine {
             if self.send(update).await.is_err() {
                 self.start(status).await;
             }
+        } else {
+            self.restart_if_needed(status).await;
         }
     }
 
@@ -686,4 +688,18 @@ impl StateMachine {
             true
         }
     }
+
+    // for states that should be running, check them and restart if needed
+    async fn restart_if_needed(&mut self, status: JobStatus) {
+        match status.state.as_str() {
+            "Running" | "Recovering" | "Rescaling" => {
+                // done() means there isn't a task running, but these states
+                // need to be advanced.
+                if self.done() {
+                    self.start(status).await;
+                }
+            }
+            _ => {}
+        }
+    }
 }
diff --git a/arroyo-controller/src/states/scheduling.rs b/arroyo-controller/src/states/scheduling.rs
index b44a9b602..4aa479acf 100644
--- a/arroyo-controller/src/states/scheduling.rs
+++ b/arroyo-controller/src/states/scheduling.rs
@@ -359,13 +359,10 @@ impl State for Scheduling {
         {
             let mut metadata = StateBackend::load_checkpoint_metadata(&ctx.config.id, epoch)
                 .await
-                .ok_or_else(|| {
+                .map_err(|err| {
                     fatal(
                         format!("Failed to restore job; checkpoint {} not found.", epoch),
-                        anyhow!(format!(
-                            "epoch {} not found for job {}",
-                            epoch, ctx.config.id
-                        )),
+                        err,
                     )
                 })?;
 
@@ -379,12 +376,25 @@ impl State for Scheduling {
                 for operator_id in &metadata.operator_ids {
                     let operator_metadata =
                         StateBackend::load_operator_metadata(&ctx.config.id, operator_id, epoch)
-                            .await;
+                            .await
+                            .map_err(|err| {
+                                fatal(
+                                    format!(
+                                "Failed to restore job; operator metadata for {} not found.",
+                                operator_id
+                            ),
+                                    err,
+                                )
+                            })?;
                     let Some(operator_metadata) = operator_metadata else {
-                        panic!(
-                            "operator metadata for {} not found for job {}",
-                            operator_id, ctx.config.id
-                        );
+                        return Err(fatal(
+                            "missing operator metadata",
+                            anyhow!(
+                                "operator metadata for {} not found for job {}",
+                                operator_id,
+                                ctx.config.id
+                            ),
+                        ));
                     };
                     if let Some(commit_data) = operator_metadata.commit_data {
                         committing_data.insert(
@@ -418,7 +428,14 @@ impl State for Scheduling {
                 }
                 committing_state = Some(CommittingState::new(id, commit_subtasks, committing_data));
             }
-            StateBackend::write_checkpoint_metadata(metadata).await;
+            StateBackend::write_checkpoint_metadata(metadata)
+                .await
+                .map_err(|err| {
+                    fatal(
+                        format!("Failed to write checkpoint metadata for epoch {}.", epoch),
+                        err,
+                    )
+                })?;
         }
 
         let assignments = compute_assignments(workers.values().collect(), ctx.program);
diff --git a/arroyo-datastream/src/lib.rs b/arroyo-datastream/src/lib.rs
index ab734b987..cb2026659 100644
--- a/arroyo-datastream/src/lib.rs
+++ b/arroyo-datastream/src/lib.rs
@@ -376,26 +376,13 @@ pub enum Operator {
         name: String,
         expression: String,
     },
-    StructToRecordBatch {
+    AsyncMapOperator {
         name: String,
+        ordered: bool,
+        function_def: String,
+        max_concurrency: u64,
+        has_context: bool,
     },
-    RecordBatchToStruct {
-        name: String,
-    },
-    ArrowValue {
-        name: String,
-        config: Vec<u8>,
-    },
-    ArrowKey {
-        name: String,
-        config: Vec<u8>,
-    },
-    RecordBatchGrpc,
-    ArrowAggregate {
-        name: String,
-        config: Vec<u8>,
-    },
-    ArrowWatermark,
 }
 
 #[derive(Clone, Encode, Decode, Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -521,13 +508,7 @@ impl Debug for Operator {
                 name,
                 expression: _,
             } => write!(f, "updating_key<{}>", name),
-            Operator::StructToRecordBatch { name } => write!(f, "struct_to_record_batch<{}>", name),
-            Operator::RecordBatchToStruct { name } => write!(f, "record_batch_to_struct<{}>", name),
-            Operator::ArrowValue { name, config: _ } => write!(f, "arrow_value<{}>", name),
-            Operator::ArrowKey { name, config: _ } => write!(f, "arrow_key<{}>", name),
-            Operator::RecordBatchGrpc => write!(f, "record_batch_sink"),
-            Operator::ArrowAggregate { name, config: _ } => write!(f, "arrow_aggregate<{}>", name),
-            Operator::ArrowWatermark => write!(f, "arrow_watermark"),
+            Operator::AsyncMapOperator { name, .. } => write!(f, "async_map<{}>", name),
         }
     }
 }
@@ -1396,7 +1377,7 @@ impl Program {
             let input = self.graph.edges_directed(idx, Direction::Incoming).next();
             let output = self.graph.edges_directed(idx, Direction::Outgoing).next();
             let body = match &node.operator {
-                Operator::ConnectorSource(c)  => {
+                Operator::ConnectorSource(c) => {
                     let out_k = parse_type(&output.unwrap().weight().key);
                     let out_t = parse_type(&output.unwrap().weight().value);
 
@@ -1406,7 +1387,7 @@ impl Program {
                         Box::new(#strukt::<#out_k, #out_t>::from_config(#config))
                     }
                 }
-                Operator::ConnectorSink(c)  => {
+                Operator::ConnectorSink(c) => {
                     // In c.operator, replace #in_k and #in_t with the actual types
                     let replaced_type = c.operator.replace("#in_k", &input.unwrap().weight().key);
                     let replaced_type = replaced_type.replace("#in_t", &input.unwrap().weight().value);
@@ -1434,15 +1415,15 @@ impl Program {
                     let out_t = parse_type(&output.unwrap().weight().value);
 
                     let agg = match agg {
-                        None => quote!{ WindowOperation::Aggregate(aggregators::vec_aggregator) },
-                        Some(WindowAgg::Count) => quote!{ WindowOperation::Aggregate(aggregators::count_aggregator) },
-                        Some(WindowAgg::Min) => quote!{ WindowOperation::Aggregate(aggregators::min_aggregator) },
-                        Some(WindowAgg::Max) => quote!{ WindowOperation::Aggregate(aggregators::max_aggregator) },
-                        Some(WindowAgg::Sum) => quote!{ WindowOperation::Aggregate(aggregators::sum_aggregator) },
+                        None => quote! { WindowOperation::Aggregate(aggregators::vec_aggregator) },
+                        Some(WindowAgg::Count) => quote! { WindowOperation::Aggregate(aggregators::count_aggregator) },
+                        Some(WindowAgg::Min) => quote! { WindowOperation::Aggregate(aggregators::min_aggregator) },
+                        Some(WindowAgg::Max) => quote! { WindowOperation::Aggregate(aggregators::max_aggregator) },
+                        Some(WindowAgg::Sum) => quote! { WindowOperation::Aggregate(aggregators::sum_aggregator) },
                         Some(WindowAgg::Expression {
-                            expression,
-                            ..
-                        }) => {
+                                 expression,
+                                 ..
+                             }) => {
                             let expr: syn::Expr = parse_str(expression).unwrap();
                             let operation = if *flatten {
                                 format_ident!("Flatten")
@@ -1455,7 +1436,7 @@ impl Program {
                                     #expr
                                 })
                             }
-                        },
+                        }
                     };
 
                     match typ {
@@ -1517,7 +1498,7 @@ impl Program {
                         }
                         WatermarkStrategy::Expression { expression } => {
                             let expr: syn::Expr = parse_str(&expression).unwrap();
-                            let watermark_function : syn::ExprClosure = parse_quote!(|record| {#expr});
+                            let watermark_function: syn::ExprClosure = parse_quote!(|record| {#expr});
                             quote! {
                                 Box::new(
                                     PeriodicWatermarkGenerator::<#in_k, #in_t>::
@@ -1579,7 +1560,7 @@ impl Program {
                     quote! {
                         Box::new(CountOperator::<#in_k, #in_t>::new())
                     }
-                },
+                }
                 Operator::Aggregate(behavior) => {
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
@@ -1594,16 +1575,16 @@ impl Program {
                     quote! {
                         Box::new(AggregateOperator::<#in_k, #in_t>::new(AggregateBehavior::#behavior))
                     }
-                },
+                }
                 Operator::FlattenOperator { name } => {
                     let k = parse_type(&output.unwrap().weight().key);
                     let t = parse_type(&output.unwrap().weight().value);
                     quote! {
                         Box::new(FlattenOperator::<#k, #t>::new(#name.to_string()))
                     }
-                },
+                }
                 Operator::ExpressionOperator { name, expression, return_type } => {
-                    let expr : syn::Expr = parse_str(expression).expect(expression);
+                    let expr: syn::Expr = parse_str(expression).expect(expression);
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
                     let out_k = parse_type(&output.unwrap().weight().key);
@@ -1617,15 +1598,15 @@ impl Program {
                                     predicate_fn: Box::new(#func),
                                 })
                             }
-                        },
+                        }
                         ExpressionReturnType::Record => {
-                    quote! {
-                        Box::new(MapOperator::<#in_k, #in_t, #out_k, #out_t> {
-                            name: #name.to_string(),
-                            map_fn: Box::new(#func),
-                        })
-                    }
-                        },
+                            quote! {
+                                Box::new(MapOperator::<#in_k, #in_t, #out_k, #out_t> {
+                                    name: #name.to_string(),
+                                    map_fn: Box::new(#func),
+                                })
+                            }
+                        }
                         ExpressionReturnType::OptionalRecord => {
                             quote! {
                                 Box::new(OptionMapOperator::<#in_k, #in_t, #out_k, #out_t> {
@@ -1633,11 +1614,11 @@ impl Program {
                                     map_fn: Box::new(#func),
                                 })
                             }
-                        },
+                        }
                     }
-                },
+                }
                 Operator::FlatMapOperator { name, expression } => {
-                    let expr : syn::Expr = parse_str(expression).expect(expression);
+                    let expr: syn::Expr = parse_str(expression).expect(expression);
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
                     let out_k = parse_type(&output.unwrap().weight().key);
@@ -1649,9 +1630,37 @@ impl Program {
                             flat_map: Box::new(#func),
                         })
                     }
-                },
+                }
+                Operator::AsyncMapOperator {
+                    name,
+                    ordered,
+                    function_def,
+                    max_concurrency,
+                    has_context
+                } => {
+                    let in_k = parse_type(&input.unwrap().weight().key);
+                    let in_t = parse_type(&input.unwrap().weight().value);
+                    let out_t = parse_type(&output.unwrap().weight().value);
+
+                    let mut context_t = quote! { EmptyContext };
+                    let mut context = quote! { EmptyContext {} };
+
+                    if *has_context {
+                        context_t = quote! { udfs::Context };
+                        context = quote! { udfs::Context::new() };
+                    }
+
+                    let udf_wrapper : syn::Expr = parse_str(function_def).unwrap();
+
+                    quote! {
+                        Box::new(AsyncMapOperator::<#in_k, #in_t, #out_t, _, _, #context_t>::
+                            new(#name.to_string(), #udf_wrapper, #context, #ordered, #max_concurrency)
+                        )
+                    }
+                }
+
                 Operator::ArrayMapOperator { name, expression, return_type } => {
-                    let expr : syn::Expr = parse_str(expression).expect(expression);
+                    let expr: syn::Expr = parse_str(expression).expect(expression);
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
                     let in_t = extract_container_type("Vec", &in_t).expect("Input to aggregate is not a Vec");
@@ -1668,14 +1677,14 @@ impl Program {
                                     }
                                 }
                             }
-                        },
+                        }
                         ExpressionReturnType::Record => {
                             quote! {
                                 |record, _| {
                                     Some(#expr)
                                 }
                             }
-                        },
+                        }
                         ExpressionReturnType::OptionalRecord => {
                             quote! {
                                 |record, _| {
@@ -1691,10 +1700,11 @@ impl Program {
                             map_fn: Box::new(#func),
                         })
                     }
-                },
-                Operator::SlidingWindowAggregator(SlidingWindowAggregator{
-                    width,slide,aggregator,bin_merger,
-                    in_memory_add,in_memory_remove,bin_type,mem_type}) => {
+                }
+                Operator::SlidingWindowAggregator(SlidingWindowAggregator {
+                                                      width, slide, aggregator, bin_merger,
+                                                      in_memory_add, in_memory_remove, bin_type, mem_type
+                                                  }) => {
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
                     let out_t = parse_type(&output.unwrap().weight().value);
@@ -1707,7 +1717,7 @@ impl Program {
                     let in_memory_add: syn::ExprClosure = parse_str(in_memory_add).unwrap();
                     let in_memory_remove: syn::ExprClosure = parse_str(in_memory_remove).unwrap();
 
-                    quote!{
+                    quote! {
                         Box::new(arroyo_worker::operators::aggregating_window::AggregatingWindowFunc::<#in_k, #in_t, #bin_t, #mem_t, #out_t>::
                             new(#width,
                                 #slide,
@@ -1716,7 +1726,7 @@ impl Program {
                                 #in_memory_add,
                                 #in_memory_remove))
                     }
-                },
+                }
                 Operator::TumblingWindowAggregator(TumblingWindowAggregator { width, aggregator, bin_merger, bin_type }) => {
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
@@ -1725,31 +1735,31 @@ impl Program {
                     let width = duration_to_syn_expr(*width);
                     let aggregator: syn::ExprClosure = parse_str(aggregator).unwrap();
                     let bin_merger: syn::ExprClosure = parse_str(bin_merger).unwrap();
-                    quote!{
+                    quote! {
                         Box::new(arroyo_worker::operators::tumbling_aggregating_window::
                             TumblingAggregatingWindowFunc::<#in_k, #in_t, #bin_t, #out_t>::
                         new(#width,
                             #aggregator,
                             #bin_merger))
                     }
-                },
+                }
                 Operator::TumblingTopN(
-                        TumblingTopN {
-                            width,
-                            max_elements,
-                            extractor,
-                            partition_key_type,
-                            converter,
-                        },
-                    ) => {
-                        let in_k = parse_type(&input.unwrap().weight().key);
-                        let in_t = parse_type(&input.unwrap().weight().value);
-                        let out_t = parse_type(&output.unwrap().weight().value);
-                        let pk_type = parse_type(partition_key_type);
-                        let width = duration_to_syn_expr(*width);
-                        let extractor: syn::ExprClosure = parse_str(extractor).expect(extractor);
-                        let converter: syn::ExprClosure = parse_str(converter).unwrap();
-                        quote! {
+                    TumblingTopN {
+                        width,
+                        max_elements,
+                        extractor,
+                        partition_key_type,
+                        converter,
+                    },
+                ) => {
+                    let in_k = parse_type(&input.unwrap().weight().key);
+                    let in_t = parse_type(&input.unwrap().weight().value);
+                    let out_t = parse_type(&output.unwrap().weight().value);
+                    let pk_type = parse_type(partition_key_type);
+                    let width = duration_to_syn_expr(*width);
+                    let extractor: syn::ExprClosure = parse_str(extractor).expect(extractor);
+                    let converter: syn::ExprClosure = parse_str(converter).unwrap();
+                    quote! {
                             Box::new(arroyo_worker::operators::tumbling_top_n_window::
                                 TumblingTopNWindowFunc::<#in_k, #in_t, #pk_type, #out_t>::
                             new(#width,
@@ -1757,7 +1767,7 @@ impl Program {
                                 #extractor,
                                 #converter))
                         }
-                    }
+                }
 
                 Operator::SlidingAggregatingTopN(
                     SlidingAggregatingTopN {
@@ -1848,24 +1858,23 @@ impl Program {
 
                     let join_fn_name = format_ident!("{}{}", join_fn_head, join_fn_tail);
 
-                    quote!{
+                    quote! {
                         Box::new(arroyo_worker::operators::joiners::
                             #join_fn_name::<#in_k, #in_t1_inner, #in_t2_inner>(#left_expiration, #right_expiration))
                     }
-
-                },
+                }
                 Operator::UpdatingOperator { name, expression } => {
-                    let expr : syn::Expr = parse_str(expression).expect(expression);
+                    let expr: syn::Expr = parse_str(expression).expect(expression);
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
                     let out_t = parse_type(&output.unwrap().weight().value);
                     let func: syn::ExprClosure = parse_quote!(|arg| {
                         #expr});
-                    quote!{
+                    quote! {
                         Box::new(arroyo_worker::operators::OptionMapOperator::<#in_k, #in_t, #in_k, #out_t>::
                             updating_operator(#name.to_string(), #func))
                     }
-                },
+                }
                 Operator::NonWindowAggregator(NonWindowAggregator { expiration, aggregator, bin_merger, bin_type }) => {
                     let in_k = parse_type(&input.unwrap().weight().key);
                     let in_t = parse_type(&input.unwrap().weight().value);
@@ -1875,70 +1884,25 @@ impl Program {
                     let expiration = duration_to_syn_expr(*expiration);
                     let aggregator: syn::ExprClosure = parse_str(aggregator).unwrap();
                     let bin_merger: syn::ExprClosure = parse_str(bin_merger).unwrap();
-                    quote!{
+                    quote! {
                         Box::new(arroyo_worker::operators::updating_aggregate::
                             UpdatingAggregateOperator::<#in_k, #in_t, #bin_t, #out_t>::
                         new(#expiration,
                             #aggregator,
                             #bin_merger))
                     }
-                },
+                }
                 Operator::UpdatingKeyOperator { name, expression } => {
                     let updating_in_t = parse_type(&input.unwrap().weight().value);
                     let in_t = extract_container_type("UpdatingData", &updating_in_t).unwrap();
                     let out_k = parse_type(&output.unwrap().weight().key);
-                    let expr : syn::Expr = parse_str(expression).expect(expression);
+                    let expr: syn::Expr = parse_str(expression).expect(expression);
                     quote! {
                         Box::new(arroyo_worker::operators::
                             KeyMapUpdatingOperator::<#in_t, #out_k>::
                         new(#name.to_string(), #expr))
                     }
-                },
-                Operator::StructToRecordBatch { name } => {
-                    let value_builder_name = parse_type(&format!("{}RecordBatchBuilder", input.unwrap().weight().value.replace(":", "_")));
-                    quote! {
-                        Box::new(arroyo_worker::arrow::
-                            StructToRecordBatch::<#value_builder_name>::
-                        new(#name.to_string()))
-                    }
-                },
-                Operator::RecordBatchToStruct { name } => {
-                    let out_k = parse_type(&output.unwrap().weight().key);
-                    let out_t = parse_type(&output.unwrap().weight().value);
-                    quote! {
-                        Box::new(arroyo_worker::arrow::
-                            RecordBatchToStruct::<#out_k, #out_t>::new(#name.to_string()))
-                    }
-                },
-                Operator::ArrowValue { name, config } => {
-                    let hex_string = hex::encode(config);
-                    quote! {
-                        Box::new(arroyo_worker::arrow::
-                            ValueExecutionOperator::from_config(#name.to_string(), hex::decode(#hex_string).unwrap()).unwrap())
-                    }
-                },
-                Operator::ArrowKey { name, config } => {
-                    let hex_string = hex::encode(config);
-                    quote! {
-                        Box::new(arroyo_worker::arrow::
-                            KeyExecutionOperator::from_config(#name.to_string(), hex::decode(#hex_string).unwrap()).unwrap())
-                    }
-                },
-                Operator::RecordBatchGrpc => quote! {
-                    Box::new(arroyo_worker::arrow::GrpcRecordBatchSink::default())
-                },
-                Operator::ArrowAggregate { name, config } => {
-                    let hex_string = hex::encode(config);
-                    quote! {
-                        Box::new(arroyo_worker::arrow::tumbling_aggregating_window::
-                            TumblingAggregatingWindowFunc::from_config(#name.to_string(), hex::decode(#hex_string).unwrap()).unwrap())
-                    }
-                },
-                Operator::ArrowWatermark => {
-                    quote!{
-                        Box::new(arroyo_worker::arrow::DefaultTimestampWatermark{})
-                    }
-                },
+                }
             };
 
             (node.operator_id.clone(), description, body, node.parallelism)
@@ -2171,6 +2135,19 @@ impl From<Operator> for GrpcApi::operator::Operator {
             Operator::FlatMapOperator { name, expression } => {
                 GrpcOperator::FlatMapOperator(GrpcApi::FlatMapOperator { name, expression })
             }
+            Operator::AsyncMapOperator {
+                name,
+                ordered,
+                function_def,
+                max_concurrency,
+                has_context,
+            } => GrpcOperator::AsyncMapOperator(GrpcApi::AsyncMapOperator {
+                name,
+                ordered,
+                function_def,
+                max_concurrency,
+                has_context,
+            }),
             Operator::ArrayMapOperator {
                 name,
                 expression,
@@ -2282,51 +2259,6 @@ impl From<Operator> for GrpcApi::operator::Operator {
             Operator::UpdatingKeyOperator { name, expression } => {
                 GrpcOperator::UpdatingKeyOperator(GrpcApi::UpdatingKeyOperator { name, expression })
             }
-            Operator::StructToRecordBatch { name } => {
-                GrpcOperator::NamedOperator(GrpcApi::NamedOperator {
-                    name,
-                    operator: GrpcApi::OperatorName::StructToRecordBatch.into(),
-                    config: None,
-                })
-            }
-            Operator::RecordBatchToStruct { name } => {
-                GrpcOperator::NamedOperator(GrpcApi::NamedOperator {
-                    name,
-                    operator: GrpcApi::OperatorName::RecordBatchToStruct.into(),
-                    config: None,
-                })
-            }
-            Operator::ArrowValue { name, config } => {
-                GrpcOperator::NamedOperator(GrpcApi::NamedOperator {
-                    name,
-                    operator: GrpcApi::OperatorName::ArrowValuePlan.into(),
-                    config: Some(config),
-                })
-            }
-            Operator::ArrowKey { name, config } => {
-                GrpcOperator::NamedOperator(GrpcApi::NamedOperator {
-                    name,
-                    operator: GrpcApi::OperatorName::ArrowKeyPlan.into(),
-                    config: Some(config),
-                })
-            }
-            Operator::RecordBatchGrpc => GrpcOperator::NamedOperator(GrpcApi::NamedOperator {
-                name: "sink".into(),
-                operator: GrpcApi::OperatorName::RecordBatchGrpc.into(),
-                ..Default::default()
-            }),
-            Operator::ArrowAggregate { name, config } => {
-                GrpcOperator::NamedOperator(GrpcApi::NamedOperator {
-                    name,
-                    operator: GrpcApi::OperatorName::ArrowAggregate.into(),
-                    config: Some(config),
-                })
-            }
-            Operator::ArrowWatermark => GrpcOperator::NamedOperator(GrpcApi::NamedOperator {
-                name: "arrow_watermark".into(),
-                operator: GrpcApi::OperatorName::ArrowWatermark.into(),
-                config: None,
-            }),
         }
     }
 }
@@ -2524,6 +2456,19 @@ impl TryFrom<arroyo_rpc::grpc::api::Operator> for Operator {
                 GrpcOperator::FlatMapOperator(GrpcApi::FlatMapOperator { name, expression }) => {
                     Operator::FlatMapOperator { name, expression }
                 }
+                GrpcOperator::AsyncMapOperator(GrpcApi::AsyncMapOperator {
+                    name,
+                    ordered,
+                    function_def,
+                    max_concurrency,
+                    has_context,
+                }) => Operator::AsyncMapOperator {
+                    name,
+                    ordered,
+                    function_def,
+                    max_concurrency,
+                    has_context,
+                },
                 GrpcOperator::FlattenExpressionOperator(flatten_expression) => {
                     let return_type = flatten_expression.return_type().into();
                     Operator::ArrayMapOperator {
@@ -2635,32 +2580,6 @@ impl TryFrom<arroyo_rpc::grpc::api::Operator> for Operator {
                     name,
                     expression,
                 }) => Operator::UpdatingKeyOperator { name, expression },
-                GrpcOperator::NamedOperator(named_operator) => {
-                    let operator = named_operator.operator();
-                    let name = named_operator.name;
-                    match operator {
-                        GrpcApi::OperatorName::StructToRecordBatch => {
-                            Operator::StructToRecordBatch { name }
-                        }
-                        GrpcApi::OperatorName::RecordBatchToStruct => {
-                            Operator::RecordBatchToStruct { name }
-                        }
-                        GrpcApi::OperatorName::ArrowValuePlan => Operator::ArrowValue {
-                            name,
-                            config: named_operator.config.unwrap(),
-                        },
-                        GrpcApi::OperatorName::ArrowKeyPlan => Operator::ArrowKey {
-                            name,
-                            config: named_operator.config.unwrap(),
-                        },
-                        GrpcApi::OperatorName::RecordBatchGrpc => Operator::RecordBatchGrpc,
-                        GrpcApi::OperatorName::ArrowAggregate => Operator::ArrowAggregate {
-                            name,
-                            config: named_operator.config.unwrap(),
-                        },
-                        GrpcApi::OperatorName::ArrowWatermark => Operator::ArrowWatermark,
-                    }
-                }
             },
             None => bail!("unset on operator {:?}", operator),
         };
diff --git a/arroyo-df/Cargo.toml b/arroyo-df/Cargo.toml
index 25e9e3d23..bfc2dac97 100644
--- a/arroyo-df/Cargo.toml
+++ b/arroyo-df/Cargo.toml
@@ -35,7 +35,7 @@ anyhow = {version = "1.0.70", features = ["backtrace"]}
 async-trait = "0.1"
 
 proc-macro2 = "1"
-syn = {version = "2", features = ["full", "parsing"]}
+syn = {version = "2", features = ["full", "parsing", "extra-traits"]}
 tracing = "0.1.37"
 
 typify = "0.0.13"
@@ -44,3 +44,4 @@ serde_json_path = "0.6.3"
 apache-avro = "0.16.0"
 prettyplease = "0.2.4"
 unicase = "2.7.0"
+toml = "0.8.8"
diff --git a/arroyo-macro/src/lib.rs b/arroyo-macro/src/lib.rs
index 653764f29..4fea6cc70 100644
--- a/arroyo-macro/src/lib.rs
+++ b/arroyo-macro/src/lib.rs
@@ -210,12 +210,14 @@ struct StreamTypesAttr {
     out_t: Option<Type>,
     timer_t: Option<Type>,
     tick_ms: Option<LitInt>,
+    futures: Option<LitStr>, // TODO: should this just be an optional bool?
 }
 
 impl Parse for StreamTypesAttr {
     fn parse(input: ParseStream) -> syn::Result<Self> {
         let mut fields = HashMap::new();
         let mut tick_ms = None;
+        let mut futures = None;
         while !input.is_empty() {
             let k: Ident = input.parse()?;
             input.parse::<Token![=]>()?;
@@ -223,6 +225,8 @@ impl Parse for StreamTypesAttr {
             let k = k.to_string();
             if k == "tick_ms" {
                 tick_ms = Some(input.parse()?);
+            } else if k == "futures" {
+                futures = Some(input.parse()?);
             } else {
                 let v: Type = input.parse()?;
 
@@ -242,6 +246,7 @@ impl Parse for StreamTypesAttr {
             out_t: fields.remove("out_t"),
             timer_t: fields.remove("timer_t"),
             tick_ms,
+            futures,
         })
     }
 }
@@ -309,6 +314,7 @@ pub fn source_fn(
         timer_t,
         stream_types_attr.tick_ms,
         item,
+        stream_types_attr.futures,
     )
 }
 
@@ -335,6 +341,7 @@ pub fn process_fn(
         timer_t,
         stream_types_attr.tick_ms,
         item,
+        stream_types_attr.futures,
     )
 }
 
@@ -367,6 +374,7 @@ pub fn co_process_fn(
         timer_t,
         stream_types_attr.tick_ms,
         item,
+        stream_types_attr.futures,
     )
 }
 
@@ -377,6 +385,7 @@ fn impl_stream_node_type(
     timer_t: Type,
     tick_ms: Option<LitInt>,
     item: proc_macro::TokenStream,
+    futures: Option<LitStr>,
 ) -> proc_macro::TokenStream {
     let mut defs = vec![];
 
@@ -442,6 +451,7 @@ fn impl_stream_node_type(
                             // do nothing
                         }
                         crate::ControlOutcome::Stop => {
+                            final_message = Some(arroyo_types::Message::Stop);
                             break;
                         }
                         crate::ControlOutcome::Finish => {
@@ -505,6 +515,21 @@ fn impl_stream_node_type(
             }
         });
 
+        let mut input_precondition = quote!(true);
+        let mut future_handler = quote!();
+        if let Some(futures) = futures {
+            let futures_ident = format_ident!("{}", futures.value());
+            future_handler = quote! {
+                Some((id, value)) = self.#futures_ident.next() => {
+                    self.handle_future(id, value, &mut ctx).await;
+                }
+            };
+
+            input_precondition = quote! {
+                self.futures.len() < self.max_concurrency as usize
+            };
+        }
+
         quote! {
             let mut counter = crate::engine::CheckpointCounter::new(in_qs.len());
             let mut closed: std::collections::HashSet<usize> = std::collections::HashSet::new();
@@ -541,7 +566,7 @@ fn impl_stream_node_type(
                             arroyo_rpc::ControlMessage::NoOp => {}
                         }
                     }
-                    p = sel.next() => {
+                    p = sel.next(), if #input_precondition => {
                         match p {
                             Some(((idx, item), s)) => {
                                 match idx / (in_partitions / #handler_count) {
@@ -556,6 +581,7 @@ fn impl_stream_node_type(
                             }
                         }
                     }
+                    #future_handler
                     #tick_case
                 }
             }
@@ -607,7 +633,7 @@ fn impl_stream_node_type(
                 let name = self.name();
                 #handle_body
 
-                Self::on_close(&mut (*self), &mut ctx).await;
+                Self::on_close(&mut (*self), &mut ctx, &final_message).await;
                 if let Some(final_message) = final_message {
                     ctx.broadcast(final_message).await;
                 }
@@ -635,10 +661,9 @@ fn impl_stream_node_type(
                 use tracing::info;
                 use tracing::trace;
                 match message {
-                    Message::Record(_) |
-                    Message::RecordBatch(_) => {
+                    Message::Record(record) => {
                         unreachable!();
-                    },
+                    }
                     Message::Barrier(t) => {
                         tracing::debug!(
                             "received barrier in {}-{}-{}-{}",
@@ -778,7 +803,7 @@ fn impl_stream_node_type(
 
     if !methods.contains("on_close") {
         defs.push(quote! {
-            async fn on_close(&mut self, ctx: &mut crate::old::Context<#out_k, #out_t>) {}
+            async fn on_close(&mut self, ctx: &mut crate::old::Context<#out_k, #out_t>, final_message: &Option<arroyo_types::Message<#out_k, #out_t>>) {}
         })
     }
 
@@ -806,7 +831,7 @@ fn impl_stream_node_type(
 
     if !methods.contains("handle_commit") {
         defs.push(quote! {
-            async fn handle_commit(&mut self, epoch: u32, commit_data: std::collections::HashMap<char, std::collections::HashMap<u32, Vec<u8>>>, ctx: &mut crate::old::Context<#out_k, #out_t>) {
+            async fn handle_commit(&mut self, epoch: u32, commit_data: std::collections::HashMap<char, std::collections::HashMap<u32, Vec<u8>>>, ctx: &mut Context<#out_k, #out_t>) {
                 tracing::warn!("default handling of commit with epoch {:?}", epoch);
             }
         })
diff --git a/arroyo-rpc/proto/api.proto b/arroyo-rpc/proto/api.proto
index ac3448571..c2ba6b037 100644
--- a/arroyo-rpc/proto/api.proto
+++ b/arroyo-rpc/proto/api.proto
@@ -91,7 +91,7 @@ message Operator {
     UpdatingOperator updating_operator = 24;
     NonWindowAggregator non_window_aggregator = 25;
     UpdatingKeyOperator updating_key_operator = 26;
-    NamedOperator named_operator = 28;
+    AsyncMapOperator async_map_operator = 28;
   }
 }
 
@@ -245,6 +245,14 @@ message FlatMapOperator {
   string expression = 2;
 }
 
+message AsyncMapOperator {
+  string name = 1;
+  bool ordered = 2;
+  string function_def = 3;
+  uint64 max_concurrency = 4;
+  bool has_context = 5;
+}
+
 message SlidingWindowAggregator {
   uint64 width_micros = 1;
   uint64 slide_micros = 2;
diff --git a/arroyo-rpc/src/lib.rs b/arroyo-rpc/src/lib.rs
index daa7d84f1..fcad6a7f8 100644
--- a/arroyo-rpc/src/lib.rs
+++ b/arroyo-rpc/src/lib.rs
@@ -289,3 +289,21 @@ impl ArroyoSchema {
             .collect()
     }
 }
+fn default_async_timeout_seconds() -> u64 {
+    10
+}
+
+fn default_async_max_concurrency() -> u64 {
+    100
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct UdfOpts {
+    #[serde(default = "bool::default")]
+    pub async_results_ordered: bool,
+    #[serde(default = "default_async_timeout_seconds")]
+    pub async_timeout_seconds: u64,
+    #[serde(default = "default_async_max_concurrency")]
+    pub async_max_concurrency: u64,
+}
diff --git a/arroyo-state/src/checkpoint_state.rs b/arroyo-state/src/checkpoint_state.rs
index eb2744021..3a7dba91c 100644
--- a/arroyo-state/src/checkpoint_state.rs
+++ b/arroyo-state/src/checkpoint_state.rs
@@ -267,7 +267,8 @@ impl CheckpointState {
                 table_checkpoint_metadata,
                 table_configs,
             })
-            .await;
+            .await
+            .expect("Should be able to write operator checkpoint metadata");
         }
         Ok(())
     }
@@ -290,7 +291,7 @@ impl CheckpointState {
                 .map(|key| key.to_string())
                 .collect(),
         })
-        .await;
+        .await?;
         Ok(())
     }
 }
diff --git a/arroyo-state/src/lib.rs b/arroyo-state/src/lib.rs
index ed389cc68..7db7b84c9 100644
--- a/arroyo-state/src/lib.rs
+++ b/arroyo-state/src/lib.rs
@@ -178,18 +178,15 @@ pub trait BackingStore {
     /// prepares a checkpoint to be loaded, e.g., by deleting future data
     async fn prepare_checkpoint_load(metadata: &CheckpointMetadata) -> Result<()>;
 
-    /// loads the latest checkpoint metadata for a given job id
-    async fn load_latest_checkpoint_metadata(job_id: &str) -> Option<CheckpointMetadata>;
-
     /// loads the checkpoint metadata for a given job id and epoch
-    async fn load_checkpoint_metadata(job_id: &str, epoch: u32) -> Option<CheckpointMetadata>;
+    async fn load_checkpoint_metadata(job_id: &str, epoch: u32) -> Result<CheckpointMetadata>;
 
     /// loads the operator checkpoint metadata for a given job id, operator id, and epoch
     async fn load_operator_metadata(
         job_id: &str,
         operator_id: &str,
         epoch: u32,
-    ) -> Option<OperatorCheckpointMetadata>;
+    ) -> Result<Option<OperatorCheckpointMetadata>>;
 
     /// creates a new instance of the BackingStore for the given task info.
     async fn new(
@@ -213,10 +210,11 @@ pub trait BackingStore {
     fn task_info(&self) -> &TaskInfo;
 
     /// writes the operator checkpoint metadata to the backing store
-    async fn write_operator_checkpoint_metadata(metadata: OperatorCheckpointMetadata);
+    async fn write_operator_checkpoint_metadata(metadata: OperatorCheckpointMetadata)
+        -> Result<()>;
 
     /// writes the checkpoint metadata to the backing store
-    async fn write_checkpoint_metadata(metadata: CheckpointMetadata);
+    async fn write_checkpoint_metadata(metadata: CheckpointMetadata) -> Result<()>;
 
     /// cleans up a checkpoint by deleting data that is no longer needed
     async fn cleanup_checkpoint(
@@ -682,7 +680,8 @@ mod test {
             commit_data: None,
             table_checkpoint_metadata: HashMap::new(),
         })
-        .await;
+        .await
+        .unwrap();
 
         let checkpoint_metadata: CheckpointMetadata = CheckpointMetadata {
             job_id: job_id.to_string(),
@@ -693,7 +692,9 @@ mod test {
             operator_ids: vec![operator_id.to_string()],
         };
 
-        ParquetBackend::write_checkpoint_metadata(checkpoint_metadata.clone()).await;
+        ParquetBackend::write_checkpoint_metadata(checkpoint_metadata.clone())
+            .await
+            .unwrap();
 
         checkpoint_metadata
     }
diff --git a/arroyo-state/src/parquet.rs b/arroyo-state/src/parquet.rs
index 40d6a58d0..72ba6b68a 100644
--- a/arroyo-state/src/parquet.rs
+++ b/arroyo-state/src/parquet.rs
@@ -117,57 +117,49 @@ impl BackingStore for ParquetBackend {
         &self.task_info
     }
 
-    async fn load_latest_checkpoint_metadata(_job_id: &str) -> Option<CheckpointMetadata> {
-        todo!()
-    }
-
     // TODO: should this be a Result, rather than an option?
-    async fn load_checkpoint_metadata(job_id: &str, epoch: u32) -> Option<CheckpointMetadata> {
-        let storage_client = get_storage_provider().await.unwrap();
+    async fn load_checkpoint_metadata(job_id: &str, epoch: u32) -> Result<CheckpointMetadata> {
+        let storage_client = get_storage_provider().await?;
         let data = storage_client
             .get(&metadata_path(&base_path(job_id, epoch)))
-            .await
-            .ok()?;
-        let metadata = CheckpointMetadata::decode(&data[..]).unwrap();
-        Some(metadata)
+            .await?;
+        let metadata = CheckpointMetadata::decode(&data[..])?;
+        Ok(metadata)
     }
 
     async fn load_operator_metadata(
         job_id: &str,
         operator_id: &str,
         epoch: u32,
-    ) -> Option<OperatorCheckpointMetadata> {
-        let storage_client = get_storage_provider().await.unwrap();
-        let data = storage_client
-            .get(&metadata_path(&operator_path(job_id, epoch, operator_id)))
-            .await
-            .ok()?;
-        Some(OperatorCheckpointMetadata::decode(&data[..]).unwrap())
+    ) -> Result<Option<OperatorCheckpointMetadata>> {
+        let storage_client = get_storage_provider().await?;
+        storage_client
+            .get_if_present(&metadata_path(&operator_path(job_id, epoch, operator_id)))
+            .await?
+            .map(|data| Ok(OperatorCheckpointMetadata::decode(&data[..])?))
+            .transpose()
     }
 
-    async fn write_operator_checkpoint_metadata(metadata: OperatorCheckpointMetadata) {
-        let storage_client = get_storage_provider().await.unwrap();
+    async fn write_operator_checkpoint_metadata(
+        metadata: OperatorCheckpointMetadata,
+    ) -> Result<()> {
+        let storage_client = get_storage_provider().await?;
         let path = metadata_path(&operator_path(
             &metadata.job_id,
             metadata.epoch,
             &metadata.operator_id,
         ));
         // TODO: propagate error
-        storage_client
-            .put(&path, metadata.encode_to_vec())
-            .await
-            .unwrap();
+        storage_client.put(&path, metadata.encode_to_vec()).await?;
+        Ok(())
     }
 
-    async fn write_checkpoint_metadata(metadata: CheckpointMetadata) {
+    async fn write_checkpoint_metadata(metadata: CheckpointMetadata) -> Result<()> {
         debug!("writing checkpoint {:?}", metadata);
-        let storage_client = get_storage_provider().await.unwrap();
+        let storage_client = get_storage_provider().await?;
         let path = metadata_path(&base_path(&metadata.job_id, metadata.epoch));
-        // TODO: propagate error
-        storage_client
-            .put(&path, metadata.encode_to_vec())
-            .await
-            .unwrap();
+        storage_client.put(&path, metadata.encode_to_vec()).await?;
+        Ok(())
     }
 
     async fn new(
@@ -206,12 +198,9 @@ impl BackingStore for ParquetBackend {
         let operator_metadata: OperatorCheckpointMetadata =
             Self::load_operator_metadata(&task_info.job_id, &task_info.operator_id, metadata.epoch)
                 .await
-                .unwrap_or_else(|| {
-                    panic!(
-                        "missing metadata for operator {}, epoch {}",
-                        task_info.operator_id, metadata.epoch
-                    )
-                });
+                // the lookup must succeed and be present.
+                .unwrap()
+                .unwrap();
         let mut current_files: HashMap<char, BTreeMap<u32, Vec<ParquetStoreData>>> = HashMap::new();
         let tables: HashMap<char, TableDescriptor> = tables
             .into_iter()
@@ -322,7 +311,7 @@ impl BackingStore for ParquetBackend {
                 .await?;
         }
         metadata.min_epoch = min_epoch;
-        Self::write_checkpoint_metadata(metadata).await;
+        Self::write_checkpoint_metadata(metadata).await?;
         Ok(())
     }
 
@@ -665,18 +654,11 @@ impl ParquetBackend {
             .parse()
             .unwrap();
 
-        let checkpoint_metadata = Self::load_checkpoint_metadata(&job_id, epoch)
-            .await
-            .unwrap_or_else(|| {
-                panic!(
-                    "missing checkpoint metadata for job {}, epoch {}",
-                    job_id, epoch
-                )
-            });
+        let checkpoint_metadata = Self::load_checkpoint_metadata(&job_id, epoch).await?;
 
         let operator_checkpoint_metadata =
             Self::load_operator_metadata(&job_id, &operator_id, epoch)
-                .await
+                .await?
                 .expect("expect operator metadata to still be present");
 
         let mut backend_data_to_drop = HashMap::new();
@@ -781,7 +763,7 @@ impl ParquetBackend {
         new_min_epoch: u32,
     ) -> Result<String> {
         let operator_metadata = Self::load_operator_metadata(&job_id, &operator_id, new_min_epoch)
-            .await
+            .await?
             .expect("expect new_min_epoch metadata to still be present");
         let paths_to_keep: HashSet<String> = operator_metadata
             .table_checkpoint_metadata
@@ -810,7 +792,7 @@ impl ParquetBackend {
 
         for epoch_to_remove in old_min_epoch..new_min_epoch {
             let Some(metadata) =
-                Self::load_operator_metadata(&job_id, &operator_id, epoch_to_remove).await
+                Self::load_operator_metadata(&job_id, &operator_id, epoch_to_remove).await?
             else {
                 continue;
             };
diff --git a/arroyo-state/src/tables/global_keyed_map.rs b/arroyo-state/src/tables/global_keyed_map.rs
index d0142d62a..5f4a74b28 100644
--- a/arroyo-state/src/tables/global_keyed_map.rs
+++ b/arroyo-state/src/tables/global_keyed_map.rs
@@ -78,6 +78,10 @@ impl<'a, K: Key, V: Data, S: BackingStore> GlobalKeyedState<'a, K, V, S> {
     pub fn get(&self, key: &K) -> Option<&V> {
         self.cache.values.get(key)
     }
+
+    pub fn get_key_values(&self) -> Vec<(&K, &V)> {
+        self.cache.values.iter().collect()
+    }
 }
 
 pub struct GlobalKeyedStateCache<K: Key, V: Data> {
diff --git a/arroyo-state/src/tables/key_time_multi_map.rs b/arroyo-state/src/tables/key_time_multi_map.rs
index dd03e43f1..fe12edac7 100644
--- a/arroyo-state/src/tables/key_time_multi_map.rs
+++ b/arroyo-state/src/tables/key_time_multi_map.rs
@@ -123,6 +123,7 @@ impl<K: Key, V: Data> KeyTimeMultiMapCache<K, V> {
             checkpoint_metadata.epoch,
         )
         .await
+        .expect("expect lookup to succeed")
         .expect("expect metadata for restoring from checkpoint");
         let min_valid_time = operator_metadata
             .min_watermark
diff --git a/arroyo-storage/src/lib.rs b/arroyo-storage/src/lib.rs
index 4cbdd524a..fef6312cb 100644
--- a/arroyo-storage/src/lib.rs
+++ b/arroyo-storage/src/lib.rs
@@ -524,6 +524,29 @@ impl StorageProvider {
         Ok(bytes)
     }
 
+    pub async fn get_if_present<P: Into<String>>(
+        &self,
+        path: P,
+    ) -> Result<Option<Bytes>, StorageError> {
+        let path: String = path.into();
+        match self
+            .object_store
+            .get(&self.qualify_path(&path.into()))
+            .await
+        {
+            Ok(obj) => {
+                let bytes = obj.bytes().await?;
+                Ok(Some(bytes))
+            }
+            Err(err) => {
+                if let object_store::Error::NotFound { .. } = &err {
+                    return Ok(None);
+                }
+                Err(err.into())
+            }
+        }
+    }
+
     pub async fn get_as_stream<P: Into<String>>(
         &self,
         path: P,
diff --git a/arroyo-types/Cargo.toml b/arroyo-types/Cargo.toml
index 19e941b5d..d75e7f1c8 100644
--- a/arroyo-types/Cargo.toml
+++ b/arroyo-types/Cargo.toml
@@ -9,3 +9,4 @@ chrono = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 arrow = { workspace = true }
 arrow-array = { workspace = true }
+async-trait = "0.1.74"
diff --git a/arroyo-types/src/lib.rs b/arroyo-types/src/lib.rs
index a3545b16f..579b0c18f 100644
--- a/arroyo-types/src/lib.rs
+++ b/arroyo-types/src/lib.rs
@@ -1,8 +1,6 @@
-use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef, TimeUnit};
-use arrow_array::builder::PrimitiveBuilder;
-use arrow_array::cast::AsArray;
-use arrow_array::types::TimestampNanosecondType;
-use arrow_array::{PrimitiveArray, RecordBatch, RecordBatchOptions, StructArray};
+use arrow::datatypes::{Field, SchemaRef};
+use arrow_array::RecordBatch;
+use async_trait::async_trait;
 use bincode::{config, BorrowDecode, Decode, Encode};
 use serde::ser::SerializeStruct;
 use serde::{Deserialize, Serialize};
@@ -330,7 +328,6 @@ pub enum Watermark {
 #[derive(Debug, Clone, Encode, Decode)]
 pub enum Message<K: Key, T: Data> {
     Record(Record<K, T>),
-    RecordBatch(RecordBatchData),
     Barrier(CheckpointBarrier),
     Watermark(Watermark),
     Stop,
@@ -372,7 +369,7 @@ impl<K: Key, T: Data> Message<K, T> {
     }
 }
 
-#[derive(Debug, Clone, Encode, Decode)]
+#[derive(Debug, Clone, Encode, Decode, PartialEq, Eq)]
 pub struct Record<K: Key, T: Data> {
     pub timestamp: SystemTime,
     pub key: Option<K>,
@@ -1081,3 +1078,9 @@ mod tests {
         );
     }
 }
+
+#[async_trait]
+pub trait UdfContext: Sync {
+    async fn init(&self) {}
+    async fn close(&self) {}
+}
diff --git a/arroyo-worker/src/arrow/mod.rs b/arroyo-worker/src/arrow/mod.rs
index 14ccb0ae1..b884abe81 100644
--- a/arroyo-worker/src/arrow/mod.rs
+++ b/arroyo-worker/src/arrow/mod.rs
@@ -9,8 +9,8 @@ use arroyo_df::physical::DecodingContext;
 use arroyo_rpc::grpc::api::ConnectorOp;
 use arroyo_rpc::grpc::controller_grpc_client::ControllerGrpcClient;
 use arroyo_rpc::grpc::{api, SinkDataReq};
-use arroyo_types::from_nanos;
 use arroyo_types::to_micros;
+use arroyo_types::{from_nanos, SignalMessage};
 use datafusion::execution::context::SessionContext;
 use datafusion::physical_plan::memory::MemoryStream;
 use datafusion::physical_plan::DisplayAs;
@@ -309,7 +309,7 @@ impl ArrowOperator for GrpcRecordBatchSink {
                 .unwrap();
         }
     }
-    async fn on_close(&mut self, ctx: &mut ArrowContext) {
+    async fn on_close(&mut self, _: &Option<SignalMessage>, ctx: &mut ArrowContext) {
         self.client
             .as_mut()
             .unwrap()
diff --git a/arroyo-worker/src/connectors/filesystem/single_file/sink.rs b/arroyo-worker/src/connectors/filesystem/single_file/sink.rs
index 0973ca9b7..cff40bfb4 100644
--- a/arroyo-worker/src/connectors/filesystem/single_file/sink.rs
+++ b/arroyo-worker/src/connectors/filesystem/single_file/sink.rs
@@ -2,12 +2,16 @@ use std::{collections::HashMap, marker::PhantomData, path::Path};
 
 use anyhow::Result;
 use arrow_array::RecordBatch;
+use arroyo_formats::SchemaData;
+use arroyo_macro::{process_fn, StreamNode};
 use arroyo_rpc::{
     grpc::{api::ConnectorOp, TableConfig},
     OperatorConfig,
 };
-use arroyo_types::CheckpointBarrier;
+use arroyo_types::{CheckpointBarrier, SignalMessage};
+use arroyo_types::{Key, Message, Record};
 use async_trait::async_trait;
+use serde::Serialize;
 use tokio::{
     fs::{self, File, OpenOptions},
     io::AsyncWriteExt,
@@ -104,8 +108,10 @@ impl ArrowOperator for FileSink {
         }
     }
 
-    async fn on_close(&mut self, _ctx: &mut ArrowContext) {
-        self.file.as_mut().unwrap().flush().await.unwrap();
+    async fn on_close(&mut self, final_message: &Option<SignalMessage>, _ctx: &mut ArrowContext) {
+        if let Some(SignalMessage::EndOfData) = final_message {
+            self.file.as_mut().unwrap().flush().await.unwrap();
+        }
     }
 
     async fn handle_checkpoint(&mut self, b: CheckpointBarrier, ctx: &mut ArrowContext) {
diff --git a/arroyo-worker/src/connectors/kafka/mod.rs b/arroyo-worker/src/connectors/kafka/mod.rs
index 99394523e..f7d40e99d 100644
--- a/arroyo-worker/src/connectors/kafka/mod.rs
+++ b/arroyo-worker/src/connectors/kafka/mod.rs
@@ -21,6 +21,7 @@ impl SourceOffset {
         match self {
             SourceOffset::Earliest => Offset::Beginning,
             SourceOffset::Latest => Offset::End,
+            SourceOffset::Group => Offset::Stored,
         }
     }
 }
diff --git a/arroyo-worker/src/connectors/kafka/sink/mod.rs b/arroyo-worker/src/connectors/kafka/sink/mod.rs
index 185f4a139..8fa071d31 100644
--- a/arroyo-worker/src/connectors/kafka/sink/mod.rs
+++ b/arroyo-worker/src/connectors/kafka/sink/mod.rs
@@ -1,24 +1,33 @@
-use anyhow::{anyhow, Result};
-use arrow_array::RecordBatch;
+use anyhow::Result;
+use arroyo_formats::SchemaData;
 use arroyo_rpc::formats::Format;
-use arroyo_rpc::{CheckpointEvent, ControlMessage, OperatorConfig};
+use arroyo_rpc::grpc::{
+    api, TableConfig, TableDeleteBehavior, TableDescriptor, TableWriteBehavior,
+};
+use arroyo_rpc::{CheckpointEvent, ControlMessage, ControlResp, OperatorConfig};
 use arroyo_types::*;
-use async_trait::async_trait;
-use rdkafka::error::KafkaError;
-use rdkafka::ClientConfig;
 use std::collections::HashMap;
-use std::time::{Duration, SystemTime};
 
-use super::{client_configs, KafkaConfig, KafkaTable, SinkCommitMode, TableType};
+use tracing::{error, warn};
+
+use rdkafka::producer::{DeliveryFuture, FutureProducer, FutureRecord, Producer};
+use rdkafka::util::Timeout;
+
+use rdkafka::ClientConfig;
+
 use crate::engine::ArrowContext;
 use crate::operator::{ArrowOperator, ArrowOperatorConstructor, OperatorNode};
+use arrow_array::RecordBatch;
 use arroyo_formats::serialize::ArrowSerializer;
 use arroyo_rpc::grpc::api::ConnectorOp;
-use arroyo_rpc::grpc::{api, TableConfig};
-use rdkafka::producer::{DeliveryFuture, FutureProducer, FutureRecord, Producer};
-use rdkafka::util::Timeout;
+use arroyo_types::CheckpointBarrier;
+use async_trait::async_trait;
+use rdkafka::error::KafkaError;
 use rdkafka_sys::RDKafkaErrorCode;
-use tracing::{error, warn};
+use serde::Serialize;
+use std::time::{Duration, SystemTime};
+
+use super::{client_configs, KafkaConfig, KafkaTable, SinkCommitMode, TableType};
 
 #[cfg(test)]
 mod test;
@@ -139,7 +148,7 @@ impl KafkaSinkFunc {
         Ok(())
     }
 
-    async fn flush(&mut self) {
+    async fn flush(&mut self, ctx: &mut ArrowContext) {
         self.producer
             .as_ref()
             .unwrap()
@@ -150,16 +159,16 @@ impl KafkaSinkFunc {
 
         // ensure all messages were delivered before finishing the checkpoint
         for future in self.write_futures.drain(..) {
-            match future.await.expect("Kafka producer shut down") {
-                Ok(_) => {}
-                Err((e, _)) => {
-                    panic!("Unhandled kafka error: {:?}", e);
-                }
+            if let Err((e, _)) = future.await.unwrap() {
+                ctx.error_reporter
+                    .report_error("Kafka producer shut down", e.to_string())
+                    .await;
+                panic!("Kafka producer shut down: {:?}", e);
             }
         }
     }
 
-    async fn publish(&mut self, k: Option<Vec<u8>>, v: Vec<u8>) {
+    async fn publish(&mut self, k: Option<Vec<u8>>, v: Vec<u8>, ctx: &mut ArrowContext) {
         let mut rec = {
             if let Some(k) = k.as_ref() {
                 FutureRecord::to(&self.topic).key(k).payload(&v)
@@ -178,7 +187,11 @@ impl KafkaSinkFunc {
                     rec = f;
                 }
                 Err((e, _)) => {
-                    panic!("Unhandled kafka error: {:?}", e);
+                    ctx.error_reporter
+                        .report_error("Could not write to Kafka", format!("{:?}", e))
+                        .await;
+
+                    panic!("Failed to write to kafka: {:?}", e);
                 }
             }
 
@@ -208,7 +221,7 @@ impl ArrowOperator for KafkaSinkFunc {
     }
 
     async fn handle_checkpoint(&mut self, _: CheckpointBarrier, ctx: &mut ArrowContext) {
-        self.flush().await;
+        self.flush(ctx).await;
         if let ConsistencyMode::ExactlyOnce {
             next_transaction_index,
             producer_to_complete,
@@ -230,16 +243,16 @@ impl ArrowOperator for KafkaSinkFunc {
     async fn process_batch(&mut self, batch: RecordBatch, ctx: &mut ArrowContext) {
         let values = self.serializer.serialize(&batch);
 
-        let keys = if !ctx.in_schemas[0].key_indices.is_empty() {
+        if !ctx.in_schemas[0].key_indices.is_empty() {
             let k = batch.project(&ctx.in_schemas[0].key_indices).unwrap();
 
             // TODO: we can probably batch this for better performance
             for (k, v) in self.serializer.serialize(&k).zip(values) {
-                self.publish(Some(k), v).await;
+                self.publish(Some(k), v, ctx).await;
             }
         } else {
             for v in values {
-                self.publish(None, v).await;
+                self.publish(None, v, ctx).await;
             }
         };
     }
@@ -276,7 +289,7 @@ impl ArrowOperator for KafkaSinkFunc {
                 commits_attempted += 1;
             }
         }
-        let checkpoint_event = arroyo_rpc::ControlResp::CheckpointEvent(CheckpointEvent {
+        let checkpoint_event = ControlResp::CheckpointEvent(CheckpointEvent {
             checkpoint_epoch: epoch,
             operator_id: ctx.task_info.operator_id.clone(),
             subtask_index: ctx.task_info.task_index as u32,
@@ -289,7 +302,7 @@ impl ArrowOperator for KafkaSinkFunc {
             .expect("sent commit event");
     }
 
-    async fn on_close(&mut self, ctx: &mut ArrowContext) {
+    async fn on_close(&mut self, _: &Option<SignalMessage>, ctx: &mut ArrowContext) {
         if !self.is_committing() {
             return;
         }
diff --git a/arroyo-worker/src/connectors/redis/mod.rs b/arroyo-worker/src/connectors/redis/mod.rs
index e2f86e178..644a8ffda 100644
--- a/arroyo-worker/src/connectors/redis/mod.rs
+++ b/arroyo-worker/src/connectors/redis/mod.rs
@@ -1,6 +1,80 @@
+use crate::connectors::redis::sink::GeneralConnection;
+use anyhow::anyhow;
+use arroyo_rpc::var_str::VarStr;
+use redis::aio::ConnectionManager;
+use redis::cluster::ClusterClient;
+use redis::{Client, ConnectionInfo, IntoConnectionInfo};
 use serde::{Deserialize, Serialize};
 use typify::import_types;
+
 pub mod sink;
 
-import_types!(schema = "../connector-schemas/redis/connection.json");
+import_types!(schema = "../connector-schemas/redis/connection.json",
+    convert = {
+        {type = "string", format = "var-str"} = VarStr
+    }
+);
 import_types!(schema = "../connector-schemas/redis/table.json");
+
+pub(crate) enum RedisClient {
+    Standard(Client),
+    Clustered(ClusterClient),
+}
+
+impl RedisClient {
+    pub fn new(config: &RedisConfig) -> anyhow::Result<Self> {
+        Ok(match &config.connection {
+            RedisConfigConnection::Address(address) => {
+                let info = from_address(&config, &address.0)?;
+
+                RedisClient::Standard(Client::open(info).map_err(|e| {
+                    anyhow!(
+                        "Failed to construct Redis client for {}: {:?}",
+                        address.0,
+                        e
+                    )
+                })?)
+            }
+            RedisConfigConnection::Addresses(addresses) => {
+                let infos: anyhow::Result<Vec<ConnectionInfo>> = addresses
+                    .iter()
+                    .map(|address| from_address(&config, address))
+                    .collect();
+
+                RedisClient::Clustered(
+                    ClusterClient::new(infos?).map_err(|e| {
+                        anyhow!("Failed to construct Redis Cluster client: {:?}", e)
+                    })?,
+                )
+            }
+        })
+    }
+
+    async fn get_connection(&self) -> Result<GeneralConnection, redis::RedisError> {
+        Ok(match self {
+            RedisClient::Standard(c) => {
+                GeneralConnection::Standard(ConnectionManager::new(c.clone()).await?)
+            }
+            RedisClient::Clustered(c) => {
+                GeneralConnection::Clustered(c.get_async_connection().await?)
+            }
+        })
+    }
+}
+
+fn from_address(config: &RedisConfig, address: &str) -> anyhow::Result<ConnectionInfo> {
+    let mut info: ConnectionInfo = address
+        .to_string()
+        .into_connection_info()
+        .map_err(|e| anyhow!("invalid redis address: {:?}", e))?;
+
+    if let Some(username) = &config.username {
+        info.redis.username = Some(username.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
+    }
+
+    if let Some(password) = &config.password {
+        info.redis.password = Some(password.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
+    }
+
+    Ok(info)
+}
diff --git a/arroyo-worker/src/connectors/redis/sink.rs b/arroyo-worker/src/connectors/redis/sink.rs
index 86e7d6054..ea8b0d505 100644
--- a/arroyo-worker/src/connectors/redis/sink.rs
+++ b/arroyo-worker/src/connectors/redis/sink.rs
@@ -1,15 +1,15 @@
-use crate::connectors::redis::{RedisConfig, RedisConfigConnection, RedisTable, TableType, Target};
+use crate::connectors::redis::{RedisClient, RedisConfig, RedisTable, TableType, Target};
 use crate::engine::{ErrorReporter, StreamNode};
 use crate::old::Context;
+
 use arroyo_formats::old::DataSerializer;
 use arroyo_formats::SchemaData;
 use arroyo_macro::process_fn;
 use arroyo_rpc::OperatorConfig;
 use arroyo_types::{CheckpointBarrier, Key, Record};
 use redis::aio::{ConnectionLike, ConnectionManager};
-use redis::cluster::ClusterClient;
 use redis::cluster_async::ClusterConnection;
-use redis::{Client, Cmd, Pipeline, RedisFuture};
+use redis::{Cmd, Pipeline, RedisFuture};
 use serde::Serialize;
 use serde_json::Value;
 use std::collections::HashSet;
@@ -32,7 +32,7 @@ where
 {
     serializer: DataSerializer<T>,
     table: RedisTable,
-    client: Clients,
+    client: RedisClient,
     cmd_q: Option<(Sender<u32>, Receiver<RedisCmd>)>,
 
     rx: Receiver<u32>,
@@ -62,22 +62,6 @@ enum RedisCmd {
     Flush(u32),
 }
 
-enum Clients {
-    Standard(Client),
-    Clustered(ClusterClient),
-}
-
-impl Clients {
-    async fn get_connection(&self) -> Result<GeneralConnection, redis::RedisError> {
-        Ok(match self {
-            Clients::Standard(c) => {
-                GeneralConnection::Standard(ConnectionManager::new(c.clone()).await?)
-            }
-            Clients::Clustered(c) => GeneralConnection::Clustered(c.get_async_connection().await?),
-        })
-    }
-}
-
 pub enum GeneralConnection {
     Standard(ConnectionManager),
     Clustered(ClusterConnection),
@@ -258,15 +242,7 @@ where
         let table: RedisTable =
             serde_json::from_value(config.table).expect("Invalid table config for Redis");
 
-        let client = match profile.connection {
-            RedisConfigConnection::Address(address) => {
-                Clients::Standard(Client::open(address.0).expect("invalid address"))
-            }
-            RedisConfigConnection::Addresses(addresses) => Clients::Clustered(
-                ClusterClient::new(addresses.into_iter().map(|e| e.0).collect())
-                    .expect("failed to construct cluster client"),
-            ),
-        };
+        let client = RedisClient::new(&profile).expect("Unable to construct redis client");
 
         let (tx, cmd_rx) = tokio::sync::mpsc::channel(128);
         let (cmd_tx, rx) = tokio::sync::mpsc::channel(128);
diff --git a/arroyo-worker/src/connectors/two_phase_committer.rs b/arroyo-worker/src/connectors/two_phase_committer.rs
index 1e3006278..dd7958818 100644
--- a/arroyo-worker/src/connectors/two_phase_committer.rs
+++ b/arroyo-worker/src/connectors/two_phase_committer.rs
@@ -8,7 +8,7 @@ use arroyo_rpc::{
     CheckpointEvent, ControlMessage,
 };
 use arroyo_state::tables::global_keyed_map::GlobalKeyedState;
-use arroyo_types::{Data, Key, Record, TaskInfo, Watermark};
+use arroyo_types::{Data, Key, Message, Record, TaskInfo, Watermark};
 use async_trait::async_trait;
 use bincode::config;
 use tracing::warn;
@@ -140,7 +140,11 @@ impl<K: Key, T: Data + Sync, TPC: TwoPhaseCommitter<K, T>> TwoPhaseCommitterOper
             .expect("record inserted");
     }
 
-    async fn on_close(&mut self, ctx: &mut crate::old::Context<(), ()>) {
+    async fn on_close(
+        &mut self,
+        ctx: &mut Context<(), ()>,
+        _final_message: &Option<Message<(), ()>>,
+    ) {
         if let Some(ControlMessage::Commit { epoch, commit_data }) = ctx.control_rx.recv().await {
             self.handle_commit(epoch, commit_data, ctx).await;
         } else {
@@ -159,7 +163,7 @@ impl<K: Key, T: Data + Sync, TPC: TwoPhaseCommitter<K, T>> TwoPhaseCommitterOper
     async fn handle_checkpoint(
         &mut self,
         checkpoint_barrier: &arroyo_types::CheckpointBarrier,
-        ctx: &mut crate::old::Context<(), ()>,
+        ctx: &mut Context<(), ()>,
     ) {
         let (recovery_data, pre_commits) = self
             .committer
@@ -208,7 +212,7 @@ impl<K: Key, T: Data + Sync, TPC: TwoPhaseCommitter<K, T>> TwoPhaseCommitterOper
         &mut self,
         epoch: u32,
         mut commit_data: HashMap<char, HashMap<u32, Vec<u8>>>,
-        ctx: &mut crate::old::Context<(), ()>,
+        ctx: &mut Context<(), ()>,
     ) {
         let pre_commits = match self.committer.commit_strategy() {
             CommitStrategy::PerSubtask => std::mem::take(&mut self.pre_commits),
diff --git a/arroyo-worker/src/connectors/webhook.rs b/arroyo-worker/src/connectors/webhook.rs
index e885b5709..af370b8cd 100644
--- a/arroyo-worker/src/connectors/webhook.rs
+++ b/arroyo-worker/src/connectors/webhook.rs
@@ -64,7 +64,7 @@ where
             .collect();
 
         Self {
-            url: Arc::new(table.endpoint),
+            url: Arc::new(table.endpoint.sub_env_vars().unwrap()),
             client: reqwest::ClientBuilder::new()
                 .default_headers(headers)
                 .timeout(Duration::from_secs(5))
diff --git a/arroyo-worker/src/engine.rs b/arroyo-worker/src/engine.rs
index 8f99c3e87..d4b9f2fd2 100644
--- a/arroyo-worker/src/engine.rs
+++ b/arroyo-worker/src/engine.rs
@@ -343,7 +343,8 @@ impl ArrowContext {
                     metadata.epoch,
                 )
                 .await
-                .unwrap();
+                .expect("lookup should succeed")
+                .expect("require metadata");
                 (metadata.min_watermark.map(from_micros), metadata)
             };
 
@@ -1133,9 +1134,10 @@ impl Engine {
             Some(
                 StateBackend::load_checkpoint_metadata(&self.job_id, epoch)
                     .await
-                    .unwrap_or_else(|| {
-                        panic!("failed to load checkpoint metadata for epoch {}", epoch)
-                    }),
+                    .expect(&format!(
+                        "failed to load checkpoint metadata for epoch {}",
+                        epoch
+                    )),
             )
         } else {
             None
diff --git a/arroyo-worker/src/lib.rs b/arroyo-worker/src/lib.rs
index 3e80c78bc..7611a72ce 100644
--- a/arroyo-worker/src/lib.rs
+++ b/arroyo-worker/src/lib.rs
@@ -78,12 +78,12 @@ pub enum SourceFinishType {
     Final,
 }
 
-impl From<SourceFinishType> for Option<ArrowMessage> {
+impl From<SourceFinishType> for Option<SignalMessage> {
     fn from(value: SourceFinishType) -> Self {
         match value {
-            SourceFinishType::Graceful => Some(ArrowMessage::Signal(SignalMessage::Stop)),
+            SourceFinishType::Graceful => Some(SignalMessage::Stop),
             SourceFinishType::Immediate => None,
-            SourceFinishType::Final => Some(ArrowMessage::Signal(SignalMessage::EndOfData)),
+            SourceFinishType::Final => Some(SignalMessage::EndOfData),
         }
     }
 }
diff --git a/arroyo-worker/src/old.rs b/arroyo-worker/src/old.rs
index 292ada744..8b327bae7 100644
--- a/arroyo-worker/src/old.rs
+++ b/arroyo-worker/src/old.rs
@@ -35,11 +35,6 @@ pub struct Collector<K: Key, T: Data> {
 }
 
 impl<K: Key, T: Data> Collector<K, T> {
-    pub async fn collect_record_batch(&mut self, record_batch: RecordBatch) {
-        let message: Message<K, T> = Message::RecordBatch(RecordBatchData(record_batch));
-        self.out_qs[0][0].send(&self.task_info, message).await;
-    }
-
     pub async fn collect(&mut self, record: Record<K, T>) {
         fn out_idx<K: Key>(key: &Option<K>, qs: usize) -> usize {
             let hash = if let Some(key) = &key {
@@ -139,7 +134,8 @@ impl<K: Key, T: Data> Context<K, T> {
                 )
                 .await;
                 metadata
-                    .expect("require metadata")
+                    .expect("failed to read")
+                    .expect("no metadata")
                     .min_watermark
                     .map(from_micros)
             };
@@ -273,10 +269,6 @@ impl<K: Key, T: Data> Context<K, T> {
         self.collector.collect(record).await;
     }
 
-    pub async fn collect_record_batch(&mut self, record: RecordBatch) {
-        self.collector.collect_record_batch(record).await;
-    }
-
     pub async fn broadcast(&mut self, message: Message<K, T>) {
         self.collector.broadcast(message).await;
     }
diff --git a/arroyo-worker/src/operator.rs b/arroyo-worker/src/operator.rs
index c09cd896a..3bc89ff81 100644
--- a/arroyo-worker/src/operator.rs
+++ b/arroyo-worker/src/operator.rs
@@ -266,7 +266,7 @@ impl OperatorNode {
         &mut self,
         ctx: &mut ArrowContext,
         in_qs: &mut Vec<Receiver<ArrowMessage>>,
-    ) -> Option<ArrowMessage> {
+    ) -> Option<SignalMessage> {
         match self {
             OperatorNode::Source(s) => {
                 s.on_start(ctx).await;
@@ -294,7 +294,7 @@ impl OperatorNode {
         let final_message = self.run_behavior(&mut ctx, &mut in_qs).await;
 
         if let Some(final_message) = final_message {
-            ctx.broadcast(final_message).await;
+            ctx.broadcast(ArrowMessage::Signal(final_message)).await;
         }
 
         info!(
@@ -365,7 +365,7 @@ async fn operator_run_behavior(
     this: &mut Box<dyn ArrowOperator>,
     ctx: &mut ArrowContext,
     in_qs: &mut Vec<Receiver<ArrowMessage>>,
-) -> Option<ArrowMessage> {
+) -> Option<SignalMessage> {
     this.on_start(ctx).await;
 
     let task_info = ctx.task_info.clone();
@@ -425,11 +425,11 @@ async fn operator_run_behavior(
                                         break;
                                     }
                                     ControlOutcome::Finish => {
-                                        final_message = Some(ArrowMessage::Signal(SignalMessage::EndOfData));
+                                        final_message = Some(SignalMessage::EndOfData);
                                         break;
                                     }
                                     ControlOutcome::StopAndSendStop => {
-                                        final_message = Some(ArrowMessage::Signal(SignalMessage::Stop));
+                                        final_message = Some(SignalMessage::Stop);
                                         break;
                                     }
                                 }
@@ -462,7 +462,7 @@ async fn operator_run_behavior(
             }
         }
     }
-    this.on_close(ctx).await;
+    this.on_close(&final_message, ctx).await;
     final_message
 }
 
@@ -649,7 +649,7 @@ pub trait ArrowOperator: Send + 'static {
     async fn handle_tick(&mut self, tick: u64, ctx: &mut ArrowContext) {}
 
     #[allow(unused_variables)]
-    async fn on_close(&mut self, ctx: &mut ArrowContext) {}
+    async fn on_close(&mut self, final_mesage: &Option<SignalMessage>, ctx: &mut ArrowContext) {}
 }
 
 pub fn get_timestamp_col<'a, 'b>(
diff --git a/arroyo-worker/src/operators/async_map.rs b/arroyo-worker/src/operators/async_map.rs
new file mode 100644
index 000000000..ccf3ce25f
--- /dev/null
+++ b/arroyo-worker/src/operators/async_map.rs
@@ -0,0 +1,270 @@
+use crate::old::Context;
+use arroyo_macro::{process_fn, StreamNode};
+use arroyo_rpc::grpc::TableDescriptor;
+use arroyo_types::{CheckpointBarrier, Data, Key, UdfContext};
+use arroyo_types::{Message, Record};
+use async_trait::async_trait;
+use futures::stream::{FuturesOrdered, FuturesUnordered};
+use futures::StreamExt;
+use std::collections::VecDeque;
+use std::future::Future;
+use std::marker::PhantomData;
+use std::sync::Arc;
+use tokio::time::error::Elapsed;
+use tracing::{debug, info, warn};
+
+pub enum FuturesEnum<T>
+where
+    T: Future + Send + 'static,
+{
+    Ordered(FuturesOrdered<T>),
+    Unordered(FuturesUnordered<T>),
+}
+
+pub struct FuturesWrapper<T>
+where
+    T: Future + Send + 'static,
+{
+    futures: FuturesEnum<T>,
+}
+
+impl<T> FuturesWrapper<T>
+where
+    T: Future + Send + 'static,
+{
+    pub fn push_back(&mut self, f: T) {
+        match &mut self.futures {
+            FuturesEnum::Ordered(futures) => futures.push_back(f),
+            FuturesEnum::Unordered(futures) => futures.push(f),
+        }
+    }
+
+    pub async fn next(&mut self) -> Option<<T as Future>::Output> {
+        match &mut self.futures {
+            FuturesEnum::Ordered(futures) => futures.next().await,
+            FuturesEnum::Unordered(futures) => futures.next().await,
+        }
+    }
+
+    pub fn len(&self) -> usize {
+        match &self.futures {
+            FuturesEnum::Ordered(futures) => futures.len(),
+            FuturesEnum::Unordered(futures) => futures.len(),
+        }
+    }
+    pub fn is_ordered(&self) -> bool {
+        match &self.futures {
+            FuturesEnum::Ordered(_) => true,
+            FuturesEnum::Unordered(_) => false,
+        }
+    }
+}
+
+#[derive(StreamNode)]
+pub struct AsyncMapOperator<
+    InKey: Key,
+    InT: Data,
+    OutT: Data,
+    FutureT: Future<Output = (usize, Result<OutT, Elapsed>)> + Send + 'static,
+    FnT: Fn(usize, InT, Arc<ContextT>) -> FutureT + Send + 'static,
+    ContextT: UdfContext + Send + 'static,
+> {
+    pub name: String,
+
+    pub udf: FnT,
+    pub futures: FuturesWrapper<FutureT>,
+    udf_context: Arc<ContextT>,
+    max_concurrency: u64,
+
+    next_id: usize, // i.e. inputs received so far, should start at 0
+    inputs: VecDeque<Option<Record<InKey, InT>>>,
+    watermarks: VecDeque<(usize, arroyo_types::Watermark)>,
+    _t: PhantomData<(InKey, InT, OutT, ContextT)>,
+}
+
+#[process_fn(in_k = InKey, in_t = InT, out_k = InKey, out_t = OutT, futures = "futures")]
+impl<
+        InKey: Key,
+        InT: Data,
+        OutT: Data,
+        FutureT: Future<Output = (usize, Result<OutT, Elapsed>)> + Send + 'static,
+        FnT: Fn(usize, InT, Arc<ContextT>) -> FutureT + Send + 'static,
+        ContextT: UdfContext + Send + 'static,
+    > AsyncMapOperator<InKey, InT, OutT, FutureT, FnT, ContextT>
+{
+    pub fn new(
+        name: String,
+        udf: FnT,
+        context: ContextT,
+        ordered: bool,
+        max_concurrency: u64,
+    ) -> Self {
+        let futures = if ordered {
+            info!("Using ordered futures");
+            FuturesWrapper {
+                futures: FuturesEnum::Ordered(FuturesOrdered::new()),
+            }
+        } else {
+            info!("Using unordered futures");
+            FuturesWrapper {
+                futures: FuturesEnum::Unordered(FuturesUnordered::new()),
+            }
+        };
+
+        Self {
+            name,
+            udf,
+            futures,
+            udf_context: Arc::new(context),
+            max_concurrency,
+            next_id: 0,
+            inputs: VecDeque::new(),
+            watermarks: VecDeque::new(),
+            _t: PhantomData,
+        }
+    }
+
+    fn name(&self) -> String {
+        self.name.clone()
+    }
+
+    async fn on_start(&mut self, ctx: &mut Context<InKey, OutT>) {
+        self.udf_context.init().await;
+
+        let gs = ctx
+            .state
+            .get_global_keyed_state::<(usize, usize), Record<InKey, InT>>('a')
+            .await;
+
+        gs.get_key_values()
+            .into_iter()
+            .filter(|(k, _)| {
+                let (task_index, _) = k;
+                task_index % ctx.task_info.parallelism == ctx.task_info.task_index
+            })
+            .for_each(|(_, v)| {
+                self.inputs.insert(self.next_id, Some(v.clone()));
+                self.futures.push_back((self.udf)(
+                    self.next_id,
+                    v.value.clone(),
+                    self.udf_context.clone(),
+                ));
+                self.next_id += 1;
+            });
+    }
+
+    async fn process_element(
+        &mut self,
+        record: &Record<InKey, InT>,
+        _ctx: &mut Context<InKey, OutT>,
+    ) {
+        self.inputs.push_back(Some(record.clone()));
+
+        self.futures.push_back((self.udf)(
+            self.next_id,
+            record.value.clone(),
+            self.udf_context.clone(),
+        ));
+        self.next_id += 1;
+    }
+
+    async fn on_collect(&mut self, id: usize, ctx: &mut Context<InKey, OutT>) {
+        // mark the input as collected by setting it to None,
+        // then pop all the Nones from the front of the queue
+        let index = self.inputs.len() - (self.next_id - id);
+        self.inputs[index] = None;
+        while let Some(None) = self.inputs.front() {
+            self.inputs.pop_front();
+            if let Some((index, _watermark)) = self.watermarks.front() {
+                // if index is 3, then that means that the watermark came in after 3 elements.
+                // if we've read in 9 more, then next_id would be 12, and 12 - 3 = 9
+                if *index + self.inputs.len() == self.next_id {
+                    let (_index, watermark) = self.watermarks.pop_front().unwrap();
+                    ctx.broadcast(arroyo_types::Message::Watermark(watermark))
+                        .await;
+                }
+            }
+        }
+    }
+
+    async fn handle_checkpoint(&mut self, _: &CheckpointBarrier, ctx: &mut Context<InKey, OutT>) {
+        let mut gs = ctx
+            .state
+            .get_global_keyed_state::<(usize, usize), Record<InKey, InT>>('a')
+            .await;
+
+        for (i, record) in self.inputs.iter().filter_map(|x| x.clone()).enumerate() {
+            let key = (ctx.task_info.task_index, i);
+            gs.insert(key, record).await;
+        }
+    }
+
+    async fn handle_watermark(
+        &mut self,
+        watermark: arroyo_types::Watermark,
+        _ctx: &mut Context<InKey, OutT>,
+    ) {
+        self.watermarks.push_back((self.next_id, watermark));
+    }
+
+    fn tables(&self) -> Vec<TableDescriptor> {
+        vec![arroyo_state::global_table("a", "AsyncMapOperator state")]
+    }
+
+    async fn on_close(
+        &mut self,
+        ctx: &mut Context<InKey, OutT>,
+        final_message: &Option<Message<InKey, OutT>>,
+    ) {
+        if let Some(Message::EndOfData) = final_message {
+            debug!(
+                "AsyncMapOperator end of data with {} futures",
+                self.futures.len()
+            );
+            while let Some((id, result)) = self.futures.next().await {
+                self.handle_future(id, result, ctx).await;
+            }
+        }
+        self.udf_context.close().await;
+    }
+
+    async fn handle_future(
+        &mut self,
+        id: usize,
+        result: Result<OutT, Elapsed>,
+        ctx: &mut Context<InKey, OutT>,
+    ) {
+        match result {
+            Ok(value) => {
+                let index = self.inputs.len() - (self.next_id - id);
+                let input = self.inputs[index].clone().unwrap();
+                ctx.collector
+                    .collect(Record {
+                        timestamp: input.timestamp,
+                        key: input.key.clone(),
+                        value,
+                    })
+                    .await;
+                self.on_collect(id, ctx).await;
+            }
+            Err(_e) => {
+                if self.futures.is_ordered() {
+                    unimplemented!(
+                        "Ordered Async UDF timed out, currently panic to preserve ordering"
+                    );
+                }
+                warn!("Unordered Async UDF timed out, retrying");
+                self.futures.push_back((self.udf)(
+                    id,
+                    self.inputs[id].clone().unwrap().value.clone(),
+                    self.udf_context.clone(),
+                ));
+            }
+        }
+    }
+}
+
+pub struct EmptyContext {}
+
+#[async_trait]
+impl UdfContext for EmptyContext {}
diff --git a/arroyo-worker/src/operators/functions/strings.rs b/arroyo-worker/src/operators/functions/strings.rs
index 569fbe4bd..31d2112a0 100644
--- a/arroyo-worker/src/operators/functions/strings.rs
+++ b/arroyo-worker/src/operators/functions/strings.rs
@@ -187,3 +187,7 @@ pub fn rtrim(string: String, trimming_characters: String) -> String {
     let char_slice: &[char] = &chars;
     string.trim_end_matches(char_slice).to_string()
 }
+
+pub fn to_hex(string: String) -> String {
+    hex::encode(string)
+}
diff --git a/arroyo-worker/src/operators/join_with_expiration.rs b/arroyo-worker/src/operators/join_with_expiration.rs
index 91e719be0..433049543 100644
--- a/arroyo-worker/src/operators/join_with_expiration.rs
+++ b/arroyo-worker/src/operators/join_with_expiration.rs
@@ -367,22 +367,10 @@ impl<K: Key, T1: Data, T2: Data> JoinProcessor<K, T1, T2, UpdatingData<(Option<T
                         ))
                     }
                 }
-                None => {
-                    if left_count == 1 {
-                        Some((
-                            left_timestamp,
-                            UpdatingData::Update {
-                                old: (Some(left.clone()), None),
-                                new: (None, None),
-                            },
-                        ))
-                    } else {
-                        Some((
-                            left_timestamp,
-                            UpdatingData::Retract((Some(left.clone()), None)),
-                        ))
-                    }
-                }
+                None => Some((
+                    left_timestamp,
+                    UpdatingData::Retract((Some(left.clone()), None)),
+                )),
             },
         }
     }
@@ -450,22 +438,10 @@ impl<K: Key, T1: Data, T2: Data> JoinProcessor<K, T1, T2, UpdatingData<(Option<T
                         ))
                     }
                 }
-                None => {
-                    if right_count == 1 {
-                        Some((
-                            right_timestamp,
-                            UpdatingData::Update {
-                                old: (None, Some(right.clone())),
-                                new: (None, None),
-                            },
-                        ))
-                    } else {
-                        Some((
-                            right_timestamp,
-                            UpdatingData::Retract((None, Some(right.clone()))),
-                        ))
-                    }
-                }
+                None => Some((
+                    right_timestamp,
+                    UpdatingData::Retract((None, Some(right.clone()))),
+                )),
             },
         }
     }
diff --git a/arroyo-worker/src/operators/mod.rs b/arroyo-worker/src/operators/mod.rs
index 3f15d065d..1548be6ba 100644
--- a/arroyo-worker/src/operators/mod.rs
+++ b/arroyo-worker/src/operators/mod.rs
@@ -29,6 +29,7 @@ use wasmtime::{
 };
 
 pub mod aggregating_window;
+pub mod async_map;
 pub mod functions;
 pub mod join_with_expiration;
 pub mod joiners;
@@ -182,13 +183,15 @@ impl ArrowOperator for PeriodicWatermarkGenerator {
         self.state_cache = state;
     }
 
-    async fn on_close(&mut self, ctx: &mut ArrowContext) {
-        // send final watermark on close
-        ctx.collector
-            .broadcast(ArrowMessage::Signal(SignalMessage::Watermark(
-                Watermark::EventTime(from_millis(u64::MAX)),
-            )))
-            .await;
+    async fn on_close(&mut self, final_message: &Option<SignalMessage>, ctx: &mut ArrowContext) {
+        if let Some(SignalMessage::EndOfData) = final_message {
+            // send final watermark on close
+            ctx.collector
+                .broadcast(ArrowMessage::Signal(SignalMessage::Watermark(
+                    Watermark::EventTime(from_millis(u64::MAX)),
+                )))
+                .await;
+        }
     }
 
     async fn process_batch(&mut self, record: RecordBatch, ctx: &mut ArrowContext) {
diff --git a/arroyo-worker/src/operators/sinks/mod.rs b/arroyo-worker/src/operators/sinks/mod.rs
index 335e5ca21..5523e0f77 100644
--- a/arroyo-worker/src/operators/sinks/mod.rs
+++ b/arroyo-worker/src/operators/sinks/mod.rs
@@ -61,7 +61,11 @@ impl<K: Key, T: Data + Serialize> GrpcSink<K, T> {
             .unwrap();
     }
 
-    async fn on_close(&mut self, ctx: &mut Context<(), ()>) {
+    async fn on_close(
+        &mut self,
+        ctx: &mut Context<(), ()>,
+        _final_message: &Option<Message<(), ()>>,
+    ) {
         self.client
             .as_mut()
             .unwrap()
diff --git a/build_dir/pipeline/Cargo.toml b/build_dir/pipeline/Cargo.toml
index 9317ae2bf..6e58efc47 100644
--- a/build_dir/pipeline/Cargo.toml
+++ b/build_dir/pipeline/Cargo.toml
@@ -23,6 +23,7 @@ arrow-array = { workspace = true}
 arroyo-types = { path = "../../arroyo-types" }
 arroyo-worker = { path = "../../arroyo-worker" }
 arroyo-formats = { path = "../../arroyo-formats" }
+tokio = "1.35.1"
 
 [package.metadata.wasm-pack.profile.release]
 wasm-opt = false
diff --git a/connector-schemas/kafka/table.json b/connector-schemas/kafka/table.json
index 88fafe5b5..87e4a7c9e 100644
--- a/connector-schemas/kafka/table.json
+++ b/connector-schemas/kafka/table.json
@@ -21,7 +21,8 @@
                             "description": "The offset to start reading from",
                             "enum": [
                                 "latest",
-                                "earliest"
+                                "earliest",
+                                "group"
                             ]
                         },
                         "read_mode": {
diff --git a/connector-schemas/redis/connection.json b/connector-schemas/redis/connection.json
index 751d7a40d..0851e2040 100644
--- a/connector-schemas/redis/connection.json
+++ b/connector-schemas/redis/connection.json
@@ -11,6 +11,7 @@
                     "properties": {
                         "address": {
                             "type": "string",
+                            "title": "address",
                             "description": "The address of your Redis server",
                             "examples": ["redis://localhost:6379"],
                             "pattern": "^redis://.*$"
@@ -43,8 +44,24 @@
                     "additionalProperties": false
                 }
             ]
+        },
+
+        "username": {
+            "title": "Username",
+            "type": "string",
+            "description": "The username for your redis cluster (if using auth)",
+            "format": "var-str"
+        },
+        "password": {
+            "title": "Password",
+            "type": "string",
+            "description": "The password for your redis cluster (if using auth)",
+            "format": "var-str"
         }
     },
+    "sensitive": [
+        "password"
+    ],
     "required": [
         "connection"
     ]
diff --git a/connector-schemas/webhook/table.json b/connector-schemas/webhook/table.json
index 08cfd37a9..722eb8930 100644
--- a/connector-schemas/webhook/table.json
+++ b/connector-schemas/webhook/table.json
@@ -9,7 +9,7 @@
             "examples": [
                 "https://yourdomain.com/api/v1/webhooks"
             ],
-            "format": "uri"
+            "format": "var-str"
         },
         "headers": {
             "title": "Headers",
diff --git a/docker/build_base/pipeline/Cargo.toml b/docker/build_base/pipeline/Cargo.toml
index 962d9fb21..44921df5f 100644
--- a/docker/build_base/pipeline/Cargo.toml
+++ b/docker/build_base/pipeline/Cargo.toml
@@ -23,6 +23,7 @@ arrow-array = { workspace = true}
 arroyo-types = { path = "/opt/arroyo/src/arroyo-types" }
 arroyo-worker = { path = "/opt/arroyo/src/arroyo-worker", features = ["kafka-sasl"] }
 arroyo-formats = { path = "/opt/arroyo/src/arroyo-formats" }
+tokio = { version = "1", features = ["full", "tracing"] }
 
 [package.metadata.wasm-pack.profile.release]
 wasm-opt = false
diff --git a/docker/build_base/pipeline/src/main.rs b/docker/build_base/pipeline/src/main.rs
index 9d060bf7f..c27a5b89e 100644
--- a/docker/build_base/pipeline/src/main.rs
+++ b/docker/build_base/pipeline/src/main.rs
@@ -1,17 +1,19 @@
-use petgraph::graph::DiGraph;
-use arroyo_worker::operators::*;
 use arroyo_worker::connectors;
-use arroyo_worker::operators::sinks::*;
-use arroyo_worker::operators::sinks;
+use arroyo_worker::engine::{Program, SubtaskNode};
 use arroyo_worker::operators::joins::*;
+use arroyo_worker::operators::sinks;
+use arroyo_worker::operators::sinks::*;
 use arroyo_worker::operators::windows::*;
-use arroyo_worker::engine::{Program, SubtaskNode};
+use arroyo_worker::operators::*;
 use arroyo_worker::{LogicalEdge, LogicalNode};
-use types::*;
 use chrono;
-use std::time::SystemTime;
-use std::str::FromStr;
+use petgraph::graph::DiGraph;
 use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::str::FromStr;
+use std::time::SystemTime;
+use tokio::time::{timeout, Duration};
+use types::*;
 
 pub fn make_graph() -> DiGraph<LogicalNode, LogicalEdge> {
     todo!();


Test patch:
diff --git a/arroyo-sql-testing/golden_outputs/async_udf.json b/arroyo-sql-testing/golden_outputs/async_udf.json
new file mode 100644
index 000000000..d12b91919
--- /dev/null
+++ b/arroyo-sql-testing/golden_outputs/async_udf.json
@@ -0,0 +1,100 @@
+{"counter":0}
+{"counter":-2}
+{"counter":-4}
+{"counter":-6}
+{"counter":-8}
+{"counter":-10}
+{"counter":-12}
+{"counter":-14}
+{"counter":-16}
+{"counter":-18}
+{"counter":-20}
+{"counter":-22}
+{"counter":-24}
+{"counter":-26}
+{"counter":-28}
+{"counter":-30}
+{"counter":-32}
+{"counter":-34}
+{"counter":-36}
+{"counter":-38}
+{"counter":-40}
+{"counter":-42}
+{"counter":-44}
+{"counter":-46}
+{"counter":-48}
+{"counter":-50}
+{"counter":-52}
+{"counter":-54}
+{"counter":-56}
+{"counter":-58}
+{"counter":-60}
+{"counter":-62}
+{"counter":-64}
+{"counter":-66}
+{"counter":-68}
+{"counter":-70}
+{"counter":-72}
+{"counter":-74}
+{"counter":-76}
+{"counter":-78}
+{"counter":-80}
+{"counter":-82}
+{"counter":-84}
+{"counter":-86}
+{"counter":-88}
+{"counter":-90}
+{"counter":-92}
+{"counter":-94}
+{"counter":-96}
+{"counter":-98}
+{"counter":-100}
+{"counter":-102}
+{"counter":-104}
+{"counter":-106}
+{"counter":-108}
+{"counter":-110}
+{"counter":-112}
+{"counter":-114}
+{"counter":-116}
+{"counter":-118}
+{"counter":-120}
+{"counter":-122}
+{"counter":-124}
+{"counter":-126}
+{"counter":-128}
+{"counter":-130}
+{"counter":-132}
+{"counter":-134}
+{"counter":-136}
+{"counter":-138}
+{"counter":-140}
+{"counter":-142}
+{"counter":-144}
+{"counter":-146}
+{"counter":-148}
+{"counter":-150}
+{"counter":-152}
+{"counter":-154}
+{"counter":-156}
+{"counter":-158}
+{"counter":-160}
+{"counter":-162}
+{"counter":-164}
+{"counter":-166}
+{"counter":-168}
+{"counter":-170}
+{"counter":-172}
+{"counter":-174}
+{"counter":-176}
+{"counter":-178}
+{"counter":-180}
+{"counter":-182}
+{"counter":-184}
+{"counter":-186}
+{"counter":-188}
+{"counter":-190}
+{"counter":-192}
+{"counter":-194}
+{"counter":-196}
+{"counter":-198}
diff --git a/arroyo-sql-testing/src/smoke_tests.rs b/arroyo-sql-testing/src/smoke_tests.rs
index a32ab5ea8..5e2bff60b 100644
--- a/arroyo-sql-testing/src/smoke_tests.rs
+++ b/arroyo-sql-testing/src/smoke_tests.rs
@@ -815,6 +815,31 @@ SELECT double_negative(counter) FROM impulse_source",
   -2 * (x as i64)
 }"}
 
+correctness_run_codegen! {"async_udf", 10,
+"CREATE TABLE impulse_source (
+  timestamp TIMESTAMP,
+  counter bigint unsigned not null,
+  subtask_index bigint unsigned not null
+) WITH (
+  connector = 'single_file',
+  path = '$input_dir/impulse.json',
+  format = 'json',
+  type = 'source'
+);
+CREATE TABLE double_negative_udf (
+  counter bigint
+) WITH (
+  connector = 'single_file',
+  path = '$output_path',
+  format = 'json',
+  type = 'sink'
+);
+INSERT INTO double_negative_udf
+SELECT double_negative(counter) FROM impulse_source",
+"pub async fn double_negative(x: u64) -> i64 {
+  -2 * (x as i64)
+}"}
+
 // test UDAF
 correctness_run_codegen! {"udaf", 10,
 "CREATE TABLE impulse_source (
diff --git a/arroyo-worker/src/connectors/kafka/source/test.rs b/arroyo-worker/src/connectors/kafka/source/test.rs
index 9ec8f770a..2bdd42fe5 100644
--- a/arroyo-worker/src/connectors/kafka/source/test.rs
+++ b/arroyo-worker/src/connectors/kafka/source/test.rs
@@ -298,7 +298,8 @@ async fn test_kafka() {
         table_checkpoint_metadata: single_item_hash_map("k", table_metadata),
         table_configs: subtask_metadata.table_configs,
     })
-    .await;
+    .await
+    .unwrap();
 
     StateBackend::write_checkpoint_metadata(CheckpointMetadata {
         job_id: task_info.job_id.clone(),
@@ -308,7 +309,8 @@ async fn test_kafka() {
         finish_time: 0,
         operator_ids: vec![task_info.operator_id.clone()],
     })
-    .await;
+    .await
+    .unwrap();
 
     reader
         .assert_next_message_record_values(