diff --git a/Cargo.lock b/Cargo.lock
index 89bde32b2..1dc45307b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -156,6 +156,29 @@ dependencies = [
  "backtrace",
 ]
 
+[[package]]
+name = "apache-avro"
+version = "0.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ceb7c683b2f8f40970b70e39ff8be514c95b96fcb9c4af87e1ed2cb2e10801a0"
+dependencies = [
+ "digest 0.10.7",
+ "lazy_static",
+ "libflate 2.0.0",
+ "log",
+ "num-bigint",
+ "quad-rand",
+ "rand",
+ "regex-lite",
+ "serde",
+ "serde_json",
+ "strum 0.25.0",
+ "strum_macros 0.25.2",
+ "thiserror",
+ "typed-builder",
+ "uuid",
+]
+
 [[package]]
 name = "arbitrary"
 version = "1.3.0"
@@ -406,6 +429,7 @@ name = "arroyo-api"
 version = "0.7.0"
 dependencies = [
  "anyhow",
+ "apache-avro",
  "argon2",
  "arrow",
  "arrow-schema",
@@ -419,7 +443,7 @@ dependencies = [
  "async-trait",
  "axum",
  "axum-extra",
- "base64 0.21.4",
+ "base64 0.21.5",
  "bincode 2.0.0-rc.3",
  "chrono",
  "cornucopia",
@@ -636,15 +660,20 @@ name = "arroyo-rpc"
 version = "0.7.0"
 dependencies = [
  "anyhow",
+ "apache-avro",
  "arroyo-types",
+ "async-trait",
  "bincode 2.0.0-rc.3",
+ "log",
  "nanoid",
  "prost",
+ "reqwest",
  "serde",
  "serde_json",
  "tokio",
  "tonic",
  "tonic-build",
+ "tracing",
  "utoipa",
 ]
 
@@ -679,6 +708,7 @@ name = "arroyo-sql"
 version = "0.7.0"
 dependencies = [
  "anyhow",
+ "apache-avro",
  "arrow",
  "arrow-schema",
  "arroyo-connectors",
@@ -806,6 +836,7 @@ name = "arroyo-worker"
 version = "0.7.0"
 dependencies = [
  "anyhow",
+ "apache-avro",
  "arrow",
  "arrow-array",
  "arroyo-macro",
@@ -1057,9 +1088,9 @@ checksum = "ecc7ab41815b3c653ccd2978ec3255c81349336702dfdf62ee6f7069b12a3aae"
 
 [[package]]
 name = "async-trait"
-version = "0.1.73"
+version = "0.1.74"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc00ceb34980c03614e35a3a4e218276a0a824e911d07651cd0d858a51e8c0f0"
+checksum = "a66537f1bb974b254c98ed142ff995236e81b9d0fe4db0575f46612cb15eb0f9"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1392,6 +1423,7 @@ checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf"
 dependencies = [
  "async-trait",
  "axum-core",
+ "axum-macros",
  "bitflags 1.3.2",
  "bytes",
  "futures-util",
@@ -1455,6 +1487,18 @@ dependencies = [
  "tower-service",
 ]
 
+[[package]]
+name = "axum-macros"
+version = "0.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cdca6a10ecad987bda04e95606ef85a5417dcaac1a78455242d72e031e2b6b62"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.33",
+]
+
 [[package]]
 name = "backoff"
 version = "0.4.0"
@@ -1510,9 +1554,9 @@ checksum = "0ea22880d78093b0cbe17c89f64a7d457941e65759157ec6cb31a31d652b05e5"
 
 [[package]]
 name = "base64"
-version = "0.21.4"
+version = "0.21.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ba43ea6f343b788c8764558649e08df62f86c6ef251fdaeb1ffd010a9ae50a2"
+checksum = "35636a1494ede3b646cc98f74f8e62c773a38a659ebc777a2cf26b9b74171df9"
 
 [[package]]
 name = "base64ct"
@@ -1628,7 +1672,7 @@ version = "0.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f03db470b3c0213c47e978da93200259a1eb4dae2e5512cba9955e2b540a6fc6"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "bollard-stubs",
  "bytes",
  "futures-core",
@@ -2014,6 +2058,15 @@ version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e496a50fda8aacccc86d7529e2c1e0892dbd0f898a6b5645b5561b89c3210efa"
 
+[[package]]
+name = "core2"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b49ba7ef1ad6107f8824dbe97de947cbaac53c44e7f9756a1fba0d37c1eec505"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "cornucopia"
 version = "0.9.0"
@@ -2406,6 +2459,12 @@ dependencies = [
  "syn 2.0.33",
 ]
 
+[[package]]
+name = "dary_heap"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7762d17f1241643615821a8455a0b2c3e803784b058693d990b11f2dce25a0ca"
+
 [[package]]
 name = "dashmap"
 version = "5.5.3"
@@ -2559,7 +2618,7 @@ dependencies = [
  "arrow-array",
  "arrow-buffer",
  "arrow-schema",
- "base64 0.21.4",
+ "base64 0.21.5",
  "blake2",
  "blake3",
  "chrono",
@@ -3093,7 +3152,7 @@ dependencies = [
  "async-lock",
  "async-rwlock",
  "async-trait",
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "cfg-if",
  "chrono",
@@ -3144,7 +3203,7 @@ checksum = "cd0337e771c38978625941b31ebce79b708c9252744f23039b9ff0434e251e02"
 dependencies = [
  "anyhow",
  "async-trait",
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "bytesize",
  "derive_builder",
@@ -3677,7 +3736,7 @@ version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06683b93020a07e3dbcf5f8c0f6d40080d725bea7936fc01ad345c01b97dc270"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "headers-core",
  "http",
@@ -4265,7 +4324,7 @@ version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cd990069640f9db34b3b0f7a1afc62a05ffaa3be9b66aa3c313f58346df7f788"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "chrono",
  "http",
@@ -4511,7 +4570,20 @@ checksum = "5ff4ae71b685bbad2f2f391fe74f6b7659a34871c08b210fdc039e43bee07d18"
 dependencies = [
  "adler32",
  "crc32fast",
- "libflate_lz77",
+ "libflate_lz77 1.2.0",
+]
+
+[[package]]
+name = "libflate"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f7d5654ae1795afc7ff76f4365c2c8791b0feb18e8996a96adad8ffd7c3b2bf"
+dependencies = [
+ "adler32",
+ "core2",
+ "crc32fast",
+ "dary_heap",
+ "libflate_lz77 2.0.0",
 ]
 
 [[package]]
@@ -4523,6 +4595,17 @@ dependencies = [
  "rle-decode-fast",
 ]
 
+[[package]]
+name = "libflate_lz77"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be5f52fb8c451576ec6b79d3f4deb327398bc05bbdbd99021a6e77a4c855d524"
+dependencies = [
+ "core2",
+ "hashbrown 0.13.2",
+ "rle-decode-fast",
+]
+
 [[package]]
 name = "libm"
 version = "0.2.7"
@@ -5097,7 +5180,7 @@ version = "0.7.1"
 source = "git+https://github.com/ArroyoSystems/arrow-rs?branch=object_store/put_part_api#612c9484547d3274571ce932260a75dc6413d6d1"
 dependencies = [
  "async-trait",
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "chrono",
  "futures",
@@ -5340,7 +5423,7 @@ dependencies = [
  "arrow-ipc",
  "arrow-schema",
  "arrow-select",
- "base64 0.21.4",
+ "base64 0.21.5",
  "brotli",
  "bytes",
  "chrono",
@@ -5603,7 +5686,7 @@ version = "0.6.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "49b6c5ef183cd3ab4ba005f1ca64c21e8bd97ce4699cfea9e8d9a2c4958ca520"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "byteorder",
  "bytes",
  "fallible-iterator",
@@ -5867,7 +5950,7 @@ checksum = "ac8a53ce01af1087eaeee6ce7c4fbf50ea4040ab1825c0115c4bafa039644ba9"
 dependencies = [
  "json",
  "libc",
- "libflate",
+ "libflate 1.4.0",
  "log",
  "names",
  "prost",
@@ -5889,6 +5972,12 @@ dependencies = [
  "thiserror",
 ]
 
+[[package]]
+name = "quad-rand"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "658fa1faf7a4cc5f057c9ee5ef560f717ad9d8dc66d975267f709624d6e1ab88"
+
 [[package]]
 name = "quanta"
 version = "0.11.1"
@@ -6137,6 +6226,12 @@ dependencies = [
  "regex-syntax 0.7.5",
 ]
 
+[[package]]
+name = "regex-lite"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "30b661b2f27137bdbc16f00eda72866a92bb28af1753ffbd56744fb6e2e9cd8e"
+
 [[package]]
 name = "regex-syntax"
 version = "0.6.29"
@@ -6161,11 +6256,11 @@ dependencies = [
 
 [[package]]
 name = "reqwest"
-version = "0.11.20"
+version = "0.11.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3e9ad3fe7488d7e34558a2033d45a0c90b72d97b4f80705666fea71472e2e6a1"
+checksum = "046cd98826c46c2ac8ddecae268eb5c2e58628688a5fc7a2643704a73faba95b"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "encoding_rs",
  "futures-core",
@@ -6189,6 +6284,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_urlencoded",
+ "system-configuration",
  "tokio",
  "tokio-native-tls",
  "tokio-rustls 0.24.1",
@@ -6508,7 +6604,7 @@ version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2d3987094b1d07b653b7dfdc3f70ce9a1da9c51ac18c1b06b662e4f9a0e9f4b2"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
 ]
 
 [[package]]
@@ -6724,9 +6820,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.106"
+version = "1.0.107"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2cc66a619ed80bf7a0f6b17dd063a84b88f6dea1813737cf469aef1d081142c2"
+checksum = "6b420ce6e3d8bd882e9b243c6eed35dbc9a6110c9769e74b584e0d68d1f20c65"
 dependencies = [
  "indexmap 2.0.0",
  "itoa",
@@ -6847,7 +6943,7 @@ version = "3.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ca3b16a3d82c4088f343b7480a93550b3eabe1a358569c2dfe38bbcead07237"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "chrono",
  "hex",
  "indexmap 1.9.3",
@@ -7305,6 +7401,27 @@ version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160"
 
+[[package]]
+name = "system-configuration"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba3a3adc5c275d719af8cb4272ea1c4a6d668a777f37e115f6d11ddbc1c8e0e7"
+dependencies = [
+ "bitflags 1.3.2",
+ "core-foundation",
+ "system-configuration-sys",
+]
+
+[[package]]
+name = "system-configuration-sys"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a75fb188eb626b924683e3b95e3a48e63551fcfb51949de2f06a9d91dbee93c9"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
 [[package]]
 name = "target-lexicon"
 version = "0.12.11"
@@ -7728,7 +7845,7 @@ checksum = "3082666a3a6433f7f511c7192923fa1fe07c69332d3c6a2e6bb040b569199d5a"
 dependencies = [
  "async-trait",
  "axum",
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "futures-core",
  "futures-util",
@@ -7780,7 +7897,7 @@ version = "0.9.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "21b00ec4842256d1fe0a46176e2ef5bc357664c66e7d91aff5a7d43d83a65f47"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "bytes",
  "futures-core",
  "http",
@@ -7820,7 +7937,7 @@ version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61c5bb1d698276a2443e5ecfabc1008bf15a36c12e6a7176e7bf089ea9131140"
 dependencies = [
- "base64 0.21.4",
+ "base64 0.21.5",
  "bitflags 2.4.0",
  "bytes",
  "futures-core",
@@ -7854,11 +7971,10 @@ checksum = "b6bc1c9ce2b5135ac7f93c72918fc37feb872bdc6a5533a8b85eb4b86bfdae52"
 
 [[package]]
 name = "tracing"
-version = "0.1.37"
+version = "0.1.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ce8c33a8d48bd45d624a6e523445fd21ec13d3653cd51f681abf67418f54eb8"
+checksum = "c3523ab5a71916ccf420eebdf5521fcef02141234bbc0b8a49f2fdc4544364ef"
 dependencies = [
- "cfg-if",
  "log",
  "pin-project-lite",
  "tracing-attributes",
@@ -7878,9 +7994,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-attributes"
-version = "0.1.26"
+version = "0.1.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5f4f31f56159e98206da9efd823404b79b6ef3143b4a7ab76e67b1751b25a4ab"
+checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -7889,9 +8005,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-core"
-version = "0.1.31"
+version = "0.1.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0955b8137a1df6f1a2e9a37d8a6656291ff0297c1a97c24e0d8425fe2312f79a"
+checksum = "c06d3da6113f116aaee68e4d601191614c9053067f9ab7f6edbcb161237daa54"
 dependencies = [
  "once_cell",
  "valuable",
@@ -7983,6 +8099,26 @@ dependencies = [
  "static_assertions",
 ]
 
+[[package]]
+name = "typed-builder"
+version = "0.16.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34085c17941e36627a879208083e25d357243812c30e7d7387c3b954f30ade16"
+dependencies = [
+ "typed-builder-macro",
+]
+
+[[package]]
+name = "typed-builder-macro"
+version = "0.16.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f03ca4cb38206e2bef0700092660bb74d696f808514dae47fa1467cbfe26e96e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.33",
+]
+
 [[package]]
 name = "typenum"
 version = "1.16.0"
@@ -8185,6 +8321,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "88ad59a7560b41a70d191093a945f0b87bc1deeda46fb237479708a1d6b6cdfc"
 dependencies = [
  "getrandom",
+ "serde",
 ]
 
 [[package]]
@@ -8408,7 +8545,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f58ddfe801df3886feaf466d883ea37e941bcc6d841b9f644a08c7acabfe7f8"
 dependencies = [
  "anyhow",
- "base64 0.21.4",
+ "base64 0.21.5",
  "bincode 1.3.3",
  "directories-next",
  "file-per-thread-logger",
diff --git a/Cargo.toml b/Cargo.toml
index 76076f44e..be113f657 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -35,7 +35,6 @@ tonic-build = { version = "0.9" }
 tonic-web = { version = "0.9" }
 tonic-reflection = { version = "0.9" }
 arrow = { version = "46.0.0" }
-arrow-buffer = { version = "46.0.0" }
 arrow-array = { version = "46.0.0" }
 arrow-schema = { version = "46.0.0" }
 object_store = { version = "0.7.1" }
@@ -44,11 +43,6 @@ parquet = { version = "46.0.0" }
 [profile.release]
 debug = 1
 
-[profile.release.package.copy-artifacts]
-# optimize for small binary size
-strip = true
-opt-level = "z"
-
 [patch.crates-io]
 typify = { git = 'https://github.com/ArroyoSystems/typify.git', branch = 'arroyo' }
 parquet = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '46.0.0/parquet_bytes'}
@@ -56,4 +50,4 @@ arrow = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '46.0.0/par
 arrow-buffer = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '46.0.0/parquet_bytes'}
 arrow-array = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '46.0.0/parquet_bytes'}
 arrow-schema = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '46.0.0/parquet_bytes'}
-object_store = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = 'object_store/put_part_api'}
\ No newline at end of file
+object_store = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = 'object_store/put_part_api'}
diff --git a/arroyo-api/Cargo.toml b/arroyo-api/Cargo.toml
index bca408689..3eaab4871 100644
--- a/arroyo-api/Cargo.toml
+++ b/arroyo-api/Cargo.toml
@@ -37,7 +37,7 @@ petgraph = {version = "0.6", features = ["serde-1"]}
 
 http = "0.2"
 tower-http = {version = "0.4", features = ["trace", "fs", "cors", "validate-request", "auth"]}
-axum = {version = "0.6.12", features = ["headers", "tokio"]}
+axum = {version = "0.6.12", features = ["headers", "tokio", "macros"]}
 axum-extra = "0.7.4"
 thiserror = "1.0.40"
 utoipa = "3"
@@ -77,6 +77,7 @@ cornucopia_async = { version = "0.4", features = ["with-serde_json-1"] }
 jwt-simple = "0.11.4"
 uuid = "1.3.3"
 regress = "0.6.0"
+apache-avro = "0.16.0"
 
 [build-dependencies]
 cornucopia = { version = "0.9" }
diff --git a/arroyo-api/src/connection_tables.rs b/arroyo-api/src/connection_tables.rs
index 2950290a6..b13788f31 100644
--- a/arroyo-api/src/connection_tables.rs
+++ b/arroyo-api/src/connection_tables.rs
@@ -9,20 +9,24 @@ use axum_extra::extract::WithRejection;
 use cornucopia_async::GenericClient;
 use cornucopia_async::Params;
 use futures_util::stream::Stream;
-use http::StatusCode;
-use serde_json::json;
+use serde_json::{json, Value};
 use std::convert::Infallible;
 use tokio::sync::mpsc::channel;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::warn;
 
+use arroyo_connectors::kafka::{KafkaConfig, KafkaTable};
 use arroyo_connectors::{connector_for_type, ErasedConnector};
 use arroyo_rpc::api_types::connections::{
-    ConfluentSchema, ConfluentSchemaQueryParams, ConnectionProfile, ConnectionSchema,
-    ConnectionTable, ConnectionTablePost, SchemaDefinition,
+    ConnectionProfile, ConnectionSchema, ConnectionTable, ConnectionTablePost, SchemaDefinition,
 };
 use arroyo_rpc::api_types::{ConnectionTableCollection, PaginationQueryParams};
+use arroyo_rpc::formats::{AvroFormat, Format, JsonFormat};
 use arroyo_rpc::public_ids::{generate_id, IdTypes};
+use arroyo_rpc::schema_resolver::{
+    ConfluentSchemaResolver, ConfluentSchemaResponse, ConfluentSchemaType,
+};
+use arroyo_sql::avro;
 use arroyo_sql::json_schema::convert_json_schema;
 use arroyo_sql::types::{StructField, TypeDef};
 
@@ -95,8 +99,9 @@ async fn get_and_validate_connector<E: GenericClient>(
         .transpose()
         .map_err(|e| bad_request(format!("Invalid schema: {}", e)))?;
 
-    let schema = if let Some(schema) = &schema {
-        Some(expand_schema(&req.name, schema)?)
+    let schema = if let Some(schema) = schema {
+        let name = connector.name();
+        Some(expand_schema(&req.name, name, schema, &req.config, &profile_config).await?)
     } else {
         None
     };
@@ -130,13 +135,14 @@ pub(crate) async fn delete_connection_table(
         .map_err(|e| handle_delete("connection_table", "pipelines", e))?;
 
     if deleted == 0 {
-        return Err(not_found("Connection table".to_string()));
+        return Err(not_found("Connection table"));
     }
 
     Ok(())
 }
 
 /// Test a Connection Table
+#[axum::debug_handler]
 #[utoipa::path(
     post,
     path = "/v1/connection_tables/test",
@@ -254,7 +260,7 @@ pub async fn create_connection_table(
         .unwrap()
         .to_string();
 
-    if let Some(schema) = &req.schema {
+    if let Some(schema) = &schema {
         if schema.definition.is_none() {
             return Err(required_field("schema.definition"));
         }
@@ -413,31 +419,109 @@ pub(crate) async fn get_connection_tables(
 
 // attempts to fill in the SQL schema from a schema object that may just have a json-schema or
 // other source schema. schemas stored in the database should always be expanded first.
-pub(crate) fn expand_schema(
+pub(crate) async fn expand_schema(
+    name: &str,
+    connector: &str,
+    schema: ConnectionSchema,
+    table_config: &Value,
+    profile_config: &Value,
+) -> Result<ConnectionSchema, ErrorResp> {
+    let Some(format) = schema.format.as_ref() else {
+        return Ok(schema);
+    };
+
+    match format {
+        Format::Json(_) => {
+            expand_json_schema(name, connector, schema, table_config, profile_config).await
+        }
+        Format::Avro(_) => {
+            expand_avro_schema(name, connector, schema, table_config, profile_config).await
+        }
+        Format::Parquet(_) => Ok(schema),
+        Format::RawString(_) => Ok(schema),
+    }
+}
+
+async fn expand_avro_schema(
     name: &str,
-    schema: &ConnectionSchema,
+    connector: &str,
+    mut schema: ConnectionSchema,
+    table_config: &Value,
+    profile_config: &Value,
 ) -> Result<ConnectionSchema, ErrorResp> {
-    let mut schema = schema.clone();
+    if let Some(Format::Avro(AvroFormat {
+        confluent_schema_registry: true,
+        ..
+    })) = &schema.format
+    {
+        let schema_response = get_schema(connector, table_config, profile_config).await?;
+
+        if schema_response.schema_type != ConfluentSchemaType::Avro {
+            return Err(bad_request(format!(
+                "Format configured is avro, but confluent schema repository returned a {:?} schema",
+                schema_response.schema_type
+            )));
+        }
+
+        schema.definition = Some(SchemaDefinition::AvroSchema(schema_response.schema));
+    }
+
+    let Some(SchemaDefinition::AvroSchema(definition)) = schema.definition.as_ref() else {
+        return Err(bad_request("avro format requires an avro schema be set"));
+    };
+
+    if let Some(Format::Avro(format)) = &mut schema.format {
+        format.add_reader_schema(
+            apache_avro::Schema::parse_str(&definition)
+                .map_err(|e| bad_request(format!("Avro schema is invalid: {:?}", e)))?,
+        );
+    }
+
+    let fields: Result<_, String> = avro::convert_avro_schema(&name, &definition)
+        .map_err(|e| bad_request(format!("Invalid avro schema: {}", e)))?
+        .into_iter()
+        .map(|f| f.try_into())
+        .collect();
+
+    schema.fields = fields.map_err(|e| bad_request(format!("Failed to convert schema: {}", e)))?;
+
+    Ok(schema)
+}
+
+async fn expand_json_schema(
+    name: &str,
+    connector: &str,
+    mut schema: ConnectionSchema,
+    table_config: &Value,
+    profile_config: &Value,
+) -> Result<ConnectionSchema, ErrorResp> {
+    if let Some(Format::Json(JsonFormat {
+        confluent_schema_registry: true,
+        ..
+    })) = &schema.format
+    {
+        let schema_response = get_schema(connector, table_config, profile_config).await?;
+
+        if schema_response.schema_type != ConfluentSchemaType::Json {
+            return Err(bad_request(format!(
+                "Format configured is json, but confluent schema repository returned a {:?} schema",
+                schema_response.schema_type
+            )));
+        }
+
+        schema.definition = Some(SchemaDefinition::JsonSchema(schema_response.schema));
+    }
 
     if let Some(d) = &schema.definition {
         let fields = match d {
-            SchemaDefinition::JsonSchema(json) => convert_json_schema(name, &json)
+            SchemaDefinition::JsonSchema(json) => convert_json_schema(&name, &json)
                 .map_err(|e| bad_request(format!("Invalid json-schema: {}", e)))?,
-            SchemaDefinition::ProtobufSchema(_) => {
-                return Err(bad_request(
-                    "Protobuf schemas are not yet supported".to_string(),
-                ))
-            }
-            SchemaDefinition::AvroSchema(_) => {
-                return Err(bad_request(
-                    "Avro schemas are not yet supported".to_string(),
-                ))
-            }
             SchemaDefinition::RawSchema(_) => vec![StructField::new(
                 "value".to_string(),
                 None,
                 TypeDef::DataType(DataType::Utf8, false),
             )],
+            _ => return Err(bad_request("Invalid schema type for json format")),
         };
 
         let fields: Result<_, String> = fields.into_iter().map(|f| f.try_into()).collect();
@@ -449,6 +533,44 @@ pub(crate) fn expand_schema(
     Ok(schema)
 }
 
+async fn get_schema(
+    connector: &str,
+    table_config: &Value,
+    profile_config: &Value,
+) -> Result<ConfluentSchemaResponse, ErrorResp> {
+    if connector != "kafka" {
+        return Err(bad_request(
+            "confluent schema registry can only be used for Kafka connections",
+        ));
+    }
+
+    // we unwrap here because this should already have been validated
+    let profile: KafkaConfig =
+        serde_json::from_value(profile_config.clone()).expect("invalid kafka config");
+
+    let table: KafkaTable =
+        serde_json::from_value(table_config.clone()).expect("invalid kafka table");
+
+    let schema_registry = profile.schema_registry.as_ref().ok_or_else(|| {
+        bad_request("schema registry must be configured on the Kafka connection profile")
+    })?;
+
+    let resolver =
+        ConfluentSchemaResolver::new(&schema_registry.endpoint, &table.topic).map_err(|e| {
+            bad_request(format!(
+                "failed to fetch schemas from schema repository: {}",
+                e
+            ))
+        })?;
+
+    resolver.get_schema(None).await.map_err(|e| {
+        bad_request(format!(
+            "failed to fetch schemas from schema repository: {}",
+            e
+        ))
+    })
+}
+
 /// Test a Connection Schema
 #[utoipa::path(
     post,
@@ -480,110 +602,3 @@ pub(crate) async fn test_schema(
         }
     }
 }
-
-/// Get a Confluent Schema
-#[utoipa::path(
-    get,
-    path = "/v1/connection_tables/schemas/confluent",
-    tag = "connection_tables",
-    params(
-        ("topic" = String, Query, description = "Confluent topic name"),
-        ("endpoint" = String, Query, description = "Confluent schema registry endpoint"),
-    ),
-    responses(
-        (status = 200, description = "Got Confluent Schema", body = ConfluentSchema),
-    ),
-)]
-pub(crate) async fn get_confluent_schema(
-    query_params: Query<ConfluentSchemaQueryParams>,
-) -> Result<Json<ConfluentSchema>, ErrorResp> {
-    // TODO: ensure only external URLs can be hit
-    let url = format!(
-        "{}/subjects/{}-value/versions/latest",
-        query_params.endpoint, query_params.topic
-    );
-    let resp = reqwest::get(url).await.map_err(|e| {
-        warn!("Got error response from schema registry: {:?}", e);
-        match e.status() {
-            Some(StatusCode::NOT_FOUND) => bad_request(format!(
-                "Could not find value schema for topic '{}'",
-                query_params.topic
-            )),
-
-            Some(code) => bad_request(format!("Schema registry returned error: {}", code)),
-            None => {
-                warn!(
-                    "Unknown error connecting to schema registry {}: {:?}",
-                    query_params.endpoint, e
-                );
-                bad_request(format!(
-                    "Could not connect to Schema Registry at {}: unknown error",
-                    query_params.endpoint
-                ))
-            }
-        }
-    })?;
-
-    if !resp.status().is_success() {
-        let message = format!(
-            "Received an error status code from the provided endpoint: {} {}",
-            resp.status().as_u16(),
-            resp.bytes()
-                .await
-                .map(|bs| String::from_utf8_lossy(&bs).to_string())
-                .unwrap_or_else(|_| "<failed to read body>".to_string())
-        );
-        return Err(bad_request(message));
-    }
-
-    let value: serde_json::Value = resp.json().await.map_err(|e| {
-        warn!("Invalid json from schema registry: {:?}", e);
-        bad_request(format!(
-            "Schema registry returned invalid JSON: {}",
-            e.to_string()
-        ))
-    })?;
-
-    let schema_type = value
-        .get("schemaType")
-        .ok_or_else(|| {
-            bad_request(
-                "The JSON returned from this endpoint was unexpected. Please confirm that the URL is correct."
-                    .to_string(),
-            )
-        })?
-        .as_str();
-
-    if schema_type != Some("JSON") {
-        return Err(bad_request(
-            "Only JSON schema types are supported currently".to_string(),
-        ));
-    }
-
-    let schema = value
-        .get("schema")
-        .ok_or_else(|| {
-            return bad_request("Missing 'schema' field in schema registry response".to_string());
-        })?
-        .as_str()
-        .ok_or_else(|| {
-            return bad_request(
-                "The 'schema' field in the schema registry response is not a string".to_string(),
-            );
-        })?;
-
-    if let Err(e) = convert_json_schema(&query_params.topic, schema) {
-        warn!(
-            "Schema from schema registry is not valid: '{}': {}",
-            schema, e
-        );
-        return Err(bad_request(format!(
-            "Schema from schema registry is not valid: {}",
-            e
-        )));
-    }
-
-    Ok(Json(ConfluentSchema {
-        schema: schema.to_string(),
-    }))
-}
diff --git a/arroyo-api/src/jobs.rs b/arroyo-api/src/jobs.rs
index f9c627663..ce6feca91 100644
--- a/arroyo-api/src/jobs.rs
+++ b/arroyo-api/src/jobs.rs
@@ -416,7 +416,7 @@ pub async fn get_checkpoint_details(
         .await
         .map_err(log_and_map)?
         .ok_or_else(|| {
-            not_found(format!(
+            not_found(&format!(
                 "Checkpoint with epoch {} for job '{}'",
                 epoch, job_pub_id
             ))
diff --git a/arroyo-api/src/lib.rs b/arroyo-api/src/lib.rs
index 1c3f2702e..43d9f4312 100644
--- a/arroyo-api/src/lib.rs
+++ b/arroyo-api/src/lib.rs
@@ -8,8 +8,8 @@ use crate::connection_profiles::{
     __path_create_connection_profile, __path_get_connection_profiles,
 };
 use crate::connection_tables::{
-    __path_create_connection_table, __path_delete_connection_table, __path_get_confluent_schema,
-    __path_get_connection_tables, __path_test_connection_table, __path_test_schema,
+    __path_create_connection_table, __path_delete_connection_table, __path_get_connection_tables,
+    __path_test_connection_table, __path_test_schema,
 };
 use crate::connectors::__path_get_connectors;
 use crate::jobs::{
@@ -148,7 +148,6 @@ pub(crate) fn to_micros(dt: OffsetDateTime) -> u64 {
         delete_connection_table,
         test_connection_table,
         test_schema,
-        get_confluent_schema,
         get_checkpoint_details,
     ),
     components(schemas(
@@ -193,7 +192,6 @@ pub(crate) fn to_micros(dt: OffsetDateTime) -> u64 {
         PrimitiveType,
         SchemaDefinition,
         TestSourceMessage,
-        ConfluentSchema,
         JsonFormat,
         AvroFormat,
         ParquetFormat,
diff --git a/arroyo-api/src/pipelines.rs b/arroyo-api/src/pipelines.rs
index 63677a9b2..ccfe6e3c9 100644
--- a/arroyo-api/src/pipelines.rs
+++ b/arroyo-api/src/pipelines.rs
@@ -600,7 +600,7 @@ pub async fn patch_pipeline(
             .opt()
             .await
             .map_err(log_and_map)?
-            .ok_or_else(|| not_found("Job".to_string()))?;
+            .ok_or_else(|| not_found("Job"))?;
 
         let program = PipelineProgram::decode(&res.program[..]).map_err(log_and_map)?;
         let map: HashMap<String, u32> = program
@@ -629,7 +629,7 @@ pub async fn patch_pipeline(
         .map_err(log_and_map)?;
 
     if res == 0 {
-        return Err(not_found("Job".to_string()));
+        return Err(not_found("Job"));
     }
 
     let pipeline = query_pipeline_by_pub_id(&pipeline_pub_id, &client, &auth_data).await?;
@@ -683,7 +683,7 @@ pub async fn restart_pipeline(
         .map_err(log_and_map)?;
 
     if res == 0 {
-        return Err(not_found("Pipeline".to_string()));
+        return Err(not_found("Pipeline"));
     }
 
     let pipeline = query_pipeline_by_pub_id(&id, &client, &auth_data).await?;
@@ -814,7 +814,7 @@ pub async fn delete_pipeline(
         .map_err(log_and_map)?;
 
     if count != 1 {
-        return Err(not_found("Pipeline".to_string()));
+        return Err(not_found("Pipeline"));
     }
 
     Ok(())
@@ -864,7 +864,7 @@ pub async fn query_pipeline_by_pub_id(
         .await
         .map_err(log_and_map)?;
 
-    let res = pipeline.ok_or_else(|| not_found("Pipeline".to_string()))?;
+    let res = pipeline.ok_or_else(|| not_found("Pipeline"))?;
 
     res.try_into()
 }
@@ -884,7 +884,7 @@ pub async fn query_job_by_pub_id(
         .await
         .map_err(log_and_map)?;
 
-    let res: DbPipelineJob = job.ok_or_else(|| not_found("Job".to_string()))?;
+    let res: DbPipelineJob = job.ok_or_else(|| not_found("Job"))?;
 
     Ok(res.into())
 }
diff --git a/arroyo-api/src/rest.rs b/arroyo-api/src/rest.rs
index b0403a327..8da8d0741 100644
--- a/arroyo-api/src/rest.rs
+++ b/arroyo-api/src/rest.rs
@@ -19,8 +19,8 @@ use utoipa_swagger_ui::SwaggerUi;
 
 use crate::connection_profiles::{create_connection_profile, get_connection_profiles};
 use crate::connection_tables::{
-    create_connection_table, delete_connection_table, get_confluent_schema, get_connection_tables,
-    test_connection_table, test_schema,
+    create_connection_table, delete_connection_table, get_connection_tables, test_connection_table,
+    test_schema,
 };
 use crate::connectors::get_connectors;
 use crate::jobs::{
@@ -55,7 +55,7 @@ pub async fn ping() -> impl IntoResponse {
 }
 
 pub async fn api_fallback() -> impl IntoResponse {
-    not_found("Route".to_string())
+    not_found("Route")
 }
 
 pub fn create_rest_app(pool: Pool, controller_addr: &str) -> Router {
@@ -112,10 +112,6 @@ pub fn create_rest_app(pool: Pool, controller_addr: &str) -> Router {
         .route("/connection_tables", post(create_connection_table))
         .route("/connection_tables/test", post(test_connection_table))
         .route("/connection_tables/schemas/test", post(test_schema))
-        .route(
-            "/connection_tables/schemas/confluent",
-            get(get_confluent_schema),
-        )
         .route("/connection_tables/:id", delete(delete_connection_table))
         .route("/pipelines", post(post_pipeline))
         .route("/pipelines", get(get_pipelines))
diff --git a/arroyo-api/src/rest_utils.rs b/arroyo-api/src/rest_utils.rs
index e09f1127f..02822f435 100644
--- a/arroyo-api/src/rest_utils.rs
+++ b/arroyo-api/src/rest_utils.rs
@@ -76,21 +76,21 @@ pub(crate) async fn authenticate(
     cloud::authenticate(client, bearer_auth).await
 }
 
-pub(crate) fn bad_request(message: String) -> ErrorResp {
+pub(crate) fn bad_request(message: impl Into<String>) -> ErrorResp {
     ErrorResp {
         status_code: StatusCode::BAD_REQUEST,
-        message: message.to_string(),
+        message: message.into(),
     }
 }
 
-pub(crate) fn unauthorized(message: String) -> ErrorResp {
+pub(crate) fn unauthorized(message: impl Into<String>) -> ErrorResp {
     ErrorResp {
         status_code: StatusCode::UNAUTHORIZED,
-        message: message.to_string(),
+        message: message.into(),
     }
 }
 
-pub(crate) fn not_found(object: String) -> ErrorResp {
+pub(crate) fn not_found(object: &str) -> ErrorResp {
     ErrorResp {
         status_code: StatusCode::NOT_FOUND,
         message: format!("{} not found", object),
diff --git a/arroyo-connectors/src/kafka.rs b/arroyo-connectors/src/kafka.rs
index eb524d839..a5e5262fa 100644
--- a/arroyo-connectors/src/kafka.rs
+++ b/arroyo-connectors/src/kafka.rs
@@ -151,9 +151,14 @@ impl Connector for KafkaConnector {
             Some(other) => bail!("unknown auth type '{}'", other),
         };
 
+        let schema_registry = opts
+            .remove("schema_registry.endpoint")
+            .map(|endpoint| SchemaRegistry { endpoint });
+
         let connection = KafkaConfig {
             authentication: auth,
             bootstrap_servers: BootstrapServers(pull_opt("bootstrap_servers", opts)?),
+            schema_registry,
         };
 
         let typ = pull_opt("type", opts)?;
diff --git a/arroyo-console/package.json b/arroyo-console/package.json
index 0fdab3292..766e8ed86 100644
--- a/arroyo-console/package.json
+++ b/arroyo-console/package.json
@@ -9,7 +9,7 @@
     "preview": "vite preview",
     "format": "npx prettier --write src/ && npx eslint --fix --ext .js,.jsx,.ts,.tsx src",
     "check": "npx prettier --check src/ && npx eslint --ext .js,.jsx,.ts,.tsx src",
-    "openapi": "cargo build && npx openapi-typescript $(pwd)/../arroyo-openapi/api-spec.json --output $(pwd)/src/gen/api-types.ts"
+    "openapi": "cargo build --package arroyo-openapi  && npx openapi-typescript $(pwd)/../arroyo-openapi/api-spec.json --output $(pwd)/src/gen/api-types.ts"
   },
   "dependencies": {
     "@babel/core": "^7.22.5",
diff --git a/arroyo-console/src/gen/api-types.ts b/arroyo-console/src/gen/api-types.ts
index 002ff3817..9dfd605ae 100644
--- a/arroyo-console/src/gen/api-types.ts
+++ b/arroyo-console/src/gen/api-types.ts
@@ -34,13 +34,6 @@ export interface paths {
      */
     post: operations["create_connection_table"];
   };
-  "/v1/connection_tables/schemas/confluent": {
-    /**
-     * Get a Confluent Schema 
-     * @description Get a Confluent Schema
-     */
-    get: operations["get_confluent_schema"];
-  };
   "/v1/connection_tables/schemas/test": {
     /**
      * Test a Connection Schema 
@@ -183,7 +176,10 @@ export type webhooks = Record<string, never>;
 
 export interface components {
   schemas: {
-    AvroFormat: Record<string, never>;
+    AvroFormat: {
+      confluentSchemaRegistry: boolean;
+      embeddedSchema: boolean;
+    };
     Checkpoint: {
       backend: string;
       /** Format: int32 */
@@ -206,9 +202,6 @@ export interface components {
     };
     /** @enum {string} */
     CheckpointSpanType: "alignment" | "sync" | "async" | "committing";
-    ConfluentSchema: {
-      schema: string;
-    };
     ConnectionProfile: {
       config: unknown;
       connector: string;
@@ -588,28 +581,6 @@ export interface operations {
       };
     };
   };
-  /**
-   * Get a Confluent Schema 
-   * @description Get a Confluent Schema
-   */
-  get_confluent_schema: {
-    parameters: {
-      query: {
-        /** @description Confluent topic name */
-        topic: string;
-        /** @description Confluent schema registry endpoint */
-        endpoint: string;
-      };
-    };
-    responses: {
-      /** @description Got Confluent Schema */
-      200: {
-        content: {
-          "application/json": components["schemas"]["ConfluentSchema"];
-        };
-      };
-    };
-  };
   /**
    * Test a Connection Schema 
    * @description Test a Connection Schema
diff --git a/arroyo-console/src/routes/connections/ConfluentSchemaEditor.tsx b/arroyo-console/src/routes/connections/ConfluentSchemaEditor.tsx
index 7076001db..9e603dec2 100644
--- a/arroyo-console/src/routes/connections/ConfluentSchemaEditor.tsx
+++ b/arroyo-console/src/routes/connections/ConfluentSchemaEditor.tsx
@@ -1,20 +1,6 @@
-import { Dispatch, useState } from 'react';
+import { Dispatch } from 'react';
 import { CreateConnectionState } from './CreateConnection';
-import {
-  Alert,
-  AlertIcon,
-  Box,
-  Button,
-  Code,
-  FormControl,
-  FormHelperText,
-  FormLabel,
-  Heading,
-  Input,
-  Stack,
-} from '@chakra-ui/react';
-import { get } from '../../lib/data_fetching';
-import { formatError } from '../../lib/util';
+import { Button, Stack, Text } from '@chakra-ui/react';
 
 export function ConfluentSchemaEditor({
   state,
@@ -25,99 +11,13 @@ export function ConfluentSchemaEditor({
   setState: Dispatch<CreateConnectionState>;
   next: () => void;
 }) {
-  const [endpoint, setEndpoint] = useState<string>('');
-  const [schema, setSchema] = useState<string | null>(null);
-  const [error, setError] = useState<string | null>(null);
-  const [loading, setLoading] = useState<boolean>(false);
-
-  const topic = state.table.topic;
-
-  const fetchSchema = async () => {
-    setError(null);
-    setLoading(true);
-    const { data, error } = await get('/v1/connection_tables/schemas/confluent', {
-      params: {
-        query: {
-          endpoint,
-          topic,
-        },
-      },
-    });
-
-    if (error) {
-      setError(formatError(error));
-    }
-
-    if (data) {
-      setSchema(data.schema);
-      setState({
-        ...state,
-        schema: {
-          ...state.schema,
-          definition: { json_schema: data.schema },
-          fields: [],
-          format: { json: { confluentSchemaRegistry: true } },
-        },
-      });
-    }
-    setLoading(false);
-  };
-
-  let errorBox = null;
-  if (error != null) {
-    errorBox = (
-      <Box>
-        <Alert status="error">
-          <AlertIcon />
-          {error}
-        </Alert>
-      </Box>
-    );
-  }
-
-  let successBox = null;
-  if (schema != null) {
-    successBox = (
-      <Stack spacing={4}>
-        <Heading size="xs">Fetched schema for topic {topic}</Heading>
-        <Code height={300} overflowY="scroll">
-          <pre>{JSON.stringify(JSON.parse(schema), null, 2)}</pre>
-        </Code>
-      </Stack>
-    );
-  }
-
   return (
     <Stack spacing={4} maxW="lg">
-      <FormControl>
-        <FormLabel>Schema Registry Endpoint</FormLabel>
-        <Input
-          type="text"
-          placeholder="http://localhost:8081"
-          value={endpoint}
-          onChange={e => setEndpoint(e.target.value)}
-        />
-        <FormHelperText>Provide the endpoint for your Confluent Schema Registry</FormHelperText>
-      </FormControl>
-
-      {errorBox}
+      <Text>Schemas will be loaded from the configured Confluent Schema Registry</Text>
 
-      <Button
-        variant="primary"
-        isDisabled={endpoint == ''}
-        isLoading={loading}
-        onClick={fetchSchema}
-      >
-        Fetch Schema
+      <Button colorScheme="green" onClick={() => next()}>
+        Continue
       </Button>
-
-      {successBox}
-
-      {schema != null && (
-        <Button colorScheme="green" onClick={() => next()}>
-          Continue
-        </Button>
-      )}
     </Stack>
   );
 }
diff --git a/arroyo-console/src/routes/connections/ConnectionTester.tsx b/arroyo-console/src/routes/connections/ConnectionTester.tsx
index 1657ddc66..e9c7782ba 100644
--- a/arroyo-console/src/routes/connections/ConnectionTester.tsx
+++ b/arroyo-console/src/routes/connections/ConnectionTester.tsx
@@ -78,7 +78,6 @@ export function ConnectionTester({
   const onClickTest = async () => {
     setTesting(true);
     setError(null);
-    console.log('config', state);
 
     if (!testing) {
       await useConnectionTableTest(sseHandler, createRequest);
@@ -89,9 +88,10 @@ export function ConnectionTester({
   const submit = async () => {
     setError(null);
     const { error } = await post('/v1/connection_tables', { body: createRequest });
-    navigate('/connections');
     if (error) {
       setError({ title: 'Failed to create connection', body: formatError(error) });
+    } else {
+      navigate('/connections');
     }
   };
 
diff --git a/arroyo-console/src/routes/connections/DefineSchema.tsx b/arroyo-console/src/routes/connections/DefineSchema.tsx
index 1c0f4bb49..6d03bde54 100644
--- a/arroyo-console/src/routes/connections/DefineSchema.tsx
+++ b/arroyo-console/src/routes/connections/DefineSchema.tsx
@@ -11,82 +11,118 @@ import {
 } from '@chakra-ui/react';
 import React, { ChangeEvent, ChangeEventHandler, Dispatch, ReactElement, useState } from 'react';
 import { CreateConnectionState } from './CreateConnection';
-import { JsonSchemaEditor } from './JsonSchemaEditor';
-import { Connector } from '../../lib/data_fetching';
+import { SchemaEditor } from './SchemaEditor';
+import {
+  ConnectionProfile,
+  ConnectionSchema,
+  Connector,
+  useConnectionProfiles,
+} from '../../lib/data_fetching';
 import { ConfluentSchemaEditor } from './ConfluentSchemaEditor';
 import { components } from '../../gen/api-types';
 
-const JsonEditor = ({
+const SchemaFormatEditor = ({
   connector,
+  connectionProfiles,
   state,
   setState,
   next,
+  format,
 }: {
   connector: Connector;
+  connectionProfiles: Array<ConnectionProfile>;
   state: CreateConnectionState;
   setState: Dispatch<CreateConnectionState>;
   next: () => void;
+  format: 'json' | 'avro';
 }) => {
   type SchemaTypeOption = { name: string; value: string };
-  let schemaTypeOptions: SchemaTypeOption[] = [
-    { name: 'JSON Schema', value: 'json' },
-    { name: 'Unstructured JSON', value: 'unstructured' },
-  ];
-  if (connector.id == 'kafka') {
-    schemaTypeOptions.push({ name: 'Confluent Schema Registry', value: 'confluent' });
+  let schemaTypeOptions: SchemaTypeOption[] = [{ name: format + ' schema', value: 'schema' }];
+
+  if (format == 'json') {
+    schemaTypeOptions.push({ name: 'Unstructured JSON', value: 'unstructured' });
   }
 
-  const [selectedSchemaType, setSelectedSchemaType] = useState<SchemaTypeOption | undefined>(
-    undefined
-  );
+  let connectionProfile = null;
+  if (state.connectionProfileId != null) {
+    connectionProfile = connectionProfiles.find(c => c.id == state.connectionProfileId);
+  }
 
-  let editor: JSX.Element | null = null;
-  switch (selectedSchemaType?.value) {
+  if (
+    connector.id == 'kafka' &&
+    connectionProfile != null &&
+    (connectionProfile.config as any).schemaRegistry != null
+  ) {
+    schemaTypeOptions.push({ name: 'Confluent Schema Registry', value: 'confluent' });
+  }
+
+  let def_name: 'json_schema' | 'avro_schema';
+  switch (format) {
     case 'json':
-      editor = <JsonSchemaEditor state={state} setState={setState} next={next} />;
+      def_name = 'json_schema';
       break;
-    case 'confluent':
-      editor = <ConfluentSchemaEditor state={state} setState={setState} next={next} />;
-      break;
-    case 'unstructured':
-      editor = (
-        <Stack spacing={4} maxW={'lg'}>
-          <Text>
-            Connection tables configured with an unstructured JSON schema have a single{' '}
-            <Code>value</Code> column with the JSON value, which can be accessed using SQL{' '}
-            <Link href="https://doc.arroyo.dev/sql/scalar-functions#json-functions">
-              JSON functions
-            </Link>
-            .
-          </Text>
-          <Button onClick={next}>Continue</Button>
-        </Stack>
-      );
+    case 'avro':
+      def_name = 'avro_schema';
       break;
+    default:
+      throw new Error('unknown format: ' + format);
+  }
+
+  let editor;
+  let value;
+  if ((state.schema?.format![format] || {})['confluentSchemaRegistry']) {
+    editor = <ConfluentSchemaEditor state={state} setState={setState} next={next} />;
+    value = 'confluent';
+  } else if (state.schema?.format?.json?.unstructured) {
+    editor = (
+      <Stack spacing={4} maxW={'lg'}>
+        <Text>
+          Connection tables configured with an unstructured JSON schema have a single{' '}
+          <Code>value</Code> column with the JSON value, which can be accessed using SQL{' '}
+          <Link href="https://doc.arroyo.dev/sql/scalar-functions#json-functions">
+            JSON functions
+          </Link>
+          .
+        </Text>
+        <Button onClick={next}>Continue</Button>
+      </Stack>
+    );
+    value = 'unstructured';
+  } else if ((state.schema?.definition || {})[def_name] != undefined) {
+    editor = <SchemaEditor state={state} setState={setState} next={next} format={format} />;
+    value = 'schema';
+  } else {
+    editor = null;
+    value = undefined;
   }
 
   const onChange = (e: React.ChangeEvent<HTMLSelectElement>) => {
-    setSelectedSchemaType(schemaTypeOptions.find(o => o.value == e.target.value));
     switch (e.target.value) {
-      case 'json':
+      case 'schema': {
         setState({
           ...state,
           schema: {
             ...state.schema,
-            definition: { json_schema: '' },
+            // @ts-ignore
+            definition: {
+              [def_name]: '',
+            },
             fields: [],
-            format: { json: { unstructured: false, confluentSchemaRegistry: false } },
+            // @ts-ignore
+            format: { [format]: {} },
           },
         });
         break;
+      }
       case 'confluent':
         setState({
           ...state,
           schema: {
             ...state.schema,
-            definition: { json_schema: '' },
+            definition: null,
             fields: [],
-            format: { json: { unstructured: false, confluentSchemaRegistry: true } },
+            // @ts-ignore
+            format: { [format]: { confluentSchemaRegistry: true } },
           },
         });
         break;
@@ -101,6 +137,17 @@ const JsonEditor = ({
           },
         });
         break;
+      default:
+        setState({
+          ...state,
+          schema: {
+            ...state.schema,
+            definition: undefined,
+            fields: [],
+            // @ts-ignore
+            format: { [format]: {} },
+          },
+        });
     }
   };
 
@@ -108,12 +155,7 @@ const JsonEditor = ({
     <Stack spacing={8}>
       <FormControl>
         <FormLabel>Schema type</FormLabel>
-        <Select
-          maxW={'lg'}
-          placeholder="Select schema type"
-          value={selectedSchemaType?.value}
-          onChange={onChange}
-        >
+        <Select maxW={'lg'} placeholder="Select schema type" value={value} onChange={onChange}>
           {schemaTypeOptions.map(s => {
             return (
               <option key={s.value} value={s.value}>
@@ -192,11 +234,40 @@ export const DefineSchema = ({
   const [selectedFormat, setSelectedFormat] = useState<string | undefined>(undefined);
   const [selectedFraming, setSelectedFraming] = useState<string | undefined>(undefined);
 
+  let { connectionProfiles, connectionProfilesLoading } = useConnectionProfiles();
+
+  if (connectionProfilesLoading) {
+    return <></>;
+  }
+
   const formats: DataFormatOption[] = [
     {
       name: 'JSON',
       value: 'json',
-      el: <JsonEditor connector={connector} state={state} setState={setState} next={next} />,
+      el: (
+        <SchemaFormatEditor
+          connector={connector}
+          connectionProfiles={connectionProfiles!}
+          state={state}
+          setState={setState}
+          next={next}
+          format={'json'}
+        />
+      ),
+    },
+    {
+      name: 'Avro',
+      value: 'avro',
+      el: (
+        <SchemaFormatEditor
+          connector={connector}
+          connectionProfiles={connectionProfiles!}
+          state={state}
+          setState={setState}
+          next={next}
+          format={'avro'}
+        />
+      ),
     },
     {
       name: 'Raw String',
@@ -208,11 +279,6 @@ export const DefineSchema = ({
       value: 'protobuf',
       disabled: true,
     },
-    {
-      name: 'Avro (coming soon)',
-      value: 'avro',
-      disabled: true,
-    },
   ];
 
   type FramingOption = {
@@ -238,31 +304,25 @@ export const DefineSchema = ({
   ];
 
   const onFormatChange = (e: ChangeEvent<DataFormatOption>) => {
-    if (String(e.target.value) == 'json') {
-      setSelectedFormat('json');
-      setState({
-        ...state,
-        schema: {
-          ...state.schema,
-          format: {
-            json: { confluentSchemaRegistry: false },
-          },
-          fields: [],
-        },
-      });
-    } else {
-      setSelectedFormat('raw_string');
-      setState({
-        ...state,
-        schema: {
-          ...state.schema,
-          format: {
-            raw_string: {},
-          },
-          fields: [],
-        },
-      });
-    }
+    let format = String(e.target.value);
+    setSelectedFormat(format);
+
+    let schema: ConnectionSchema = {
+      ...state.schema,
+      // @ts-ignore
+      format: {},
+      fields: [],
+      // @ts-ignore
+      definition: {},
+    };
+
+    // @ts-ignore
+    schema.format[format] = {};
+
+    setState({
+      ...state,
+      schema,
+    });
   };
 
   const onFramingChange: ChangeEventHandler<HTMLSelectElement> = e => {
diff --git a/arroyo-console/src/routes/connections/JsonForm.tsx b/arroyo-console/src/routes/connections/JsonForm.tsx
index e005e1b09..0006d18c0 100644
--- a/arroyo-console/src/routes/connections/JsonForm.tsx
+++ b/arroyo-console/src/routes/connections/JsonForm.tsx
@@ -140,7 +140,7 @@ function SelectWidget({
   onChange: (e: React.ChangeEvent<any>) => void;
 }) {
   return (
-    <FormControl>
+    <FormControl isRequired={required}>
       {title && <FormLabel>{title}</FormLabel>}
       <Select
         placeholder={placeholder}
@@ -298,6 +298,7 @@ export function FormInner({
                       title={property.title || key}
                       description={property.description}
                       placeholder="Select an option"
+                      required={schema.required?.includes(key)}
                       options={property.enum.map(value => ({
                         value: value!.toString(),
                         label: value!.toString(),
@@ -344,7 +345,6 @@ export function FormInner({
                     onChange={onChange}
                   />
                 );
-                342;
               }
               case 'array': {
                 return (
@@ -401,7 +401,7 @@ export function FormInner({
                       </Stack>
                     </fieldset>
                   );
-                } else if ((values[key].properties?.length || 0) > 0) {
+                } else if (property.properties != undefined) {
                   return (
                     <fieldset key={key} style={{ border: '1px solid #888', borderRadius: '8px' }}>
                       <legend
diff --git a/arroyo-console/src/routes/connections/JsonSchemaEditor.tsx b/arroyo-console/src/routes/connections/SchemaEditor.tsx
similarity index 84%
rename from arroyo-console/src/routes/connections/JsonSchemaEditor.tsx
rename to arroyo-console/src/routes/connections/SchemaEditor.tsx
index 167f32fd2..fe0332444 100644
--- a/arroyo-console/src/routes/connections/JsonSchemaEditor.tsx
+++ b/arroyo-console/src/routes/connections/SchemaEditor.tsx
@@ -2,17 +2,19 @@ import { Dispatch, useEffect, useRef, useState } from 'react';
 import { CreateConnectionState } from './CreateConnection';
 import { Alert, AlertIcon, Box, Button, List, ListItem, Stack } from '@chakra-ui/react';
 import * as monaco from 'monaco-editor/esm/vs/editor/editor.api';
-import { post } from '../../lib/data_fetching';
+import { ConnectionSchema, post } from '../../lib/data_fetching';
 import { formatError } from '../../lib/util';
 
-export function JsonSchemaEditor({
+export function SchemaEditor({
   state,
   setState,
   next,
+  format,
 }: {
   state: CreateConnectionState;
   setState: Dispatch<CreateConnectionState>;
   next: () => void;
+  format: 'avro' | 'json';
 }) {
   const [editor, setEditor] = useState<monaco.editor.IStandaloneCodeEditor | null>(null);
   const monacoEl = useRef(null);
@@ -24,7 +26,6 @@ export function JsonSchemaEditor({
   const valid = tested == editor?.getValue() && errors?.length == 0;
 
   const testSchema = async () => {
-    console.log('testing schema');
     setTesting(true);
     setErrors(null);
     const { error } = await post('/v1/connection_tables/schemas/test', {
@@ -75,7 +76,6 @@ export function JsonSchemaEditor({
       state.schema?.format?.json?.unstructured === false
     ) {
       let e = monaco.editor.create(monacoEl.current!, {
-        // value: state.schema?.definition,
         language: 'json',
         theme: 'vs-dark',
         minimap: {
@@ -84,14 +84,23 @@ export function JsonSchemaEditor({
       });
 
       e?.getModel()?.onDidChangeContent(_ => {
+        let schema: ConnectionSchema = {
+          ...state.schema,
+          fields: [],
+          // @ts-ignore
+          format: {},
+          // @ts-ignore
+          definition: {},
+        };
+
+        // @ts-ignore
+        schema.format![format] = {};
+        // @ts-ignore
+        schema.definition![format + '_schema'] = e.getValue();
+
         setState({
           ...state,
-          schema: {
-            ...state.schema,
-            definition: { json_schema: e.getValue() },
-            fields: [],
-            format: { json: { confluentSchemaRegistry: false } },
-          },
+          schema: schema,
         });
       });
 
diff --git a/arroyo-rpc/Cargo.toml b/arroyo-rpc/Cargo.toml
index 4ffb1667d..43c9d2591 100644
--- a/arroyo-rpc/Cargo.toml
+++ b/arroyo-rpc/Cargo.toml
@@ -17,6 +17,11 @@ serde_json = "1.0"
 nanoid = "0.4"
 utoipa = "3"
 anyhow = "1.0.75"
+reqwest = { version = "0.11.22", features = ["default", "serde_json", "json"] }
+log = "0.4.20"
+tracing = "0.1.40"
+async-trait = "0.1.74"
+apache-avro = "0.16.0"
 
 [build-dependencies]
 tonic-build = { workspace = true }
diff --git a/arroyo-rpc/src/formats.rs b/arroyo-rpc/src/formats.rs
index 7114ac545..bfc2b90de 100644
--- a/arroyo-rpc/src/formats.rs
+++ b/arroyo-rpc/src/formats.rs
@@ -1,5 +1,7 @@
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use std::cmp::Ordering;
 use std::collections::HashMap;
+use std::hash::{Hash, Hasher};
 use std::str::FromStr;
 use utoipa::ToSchema;
 
@@ -91,9 +93,112 @@ impl JsonFormat {
 #[serde(rename_all = "camelCase")]
 pub struct RawStringFormat {}
 
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, Hash, PartialOrd, ToSchema)]
+pub struct ConfluentSchemaRegistryConfig {
+    endpoint: String,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct SerializableAvroSchema(pub apache_avro::Schema);
+
+impl Eq for SerializableAvroSchema {}
+
+impl Hash for SerializableAvroSchema {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.0.canonical_form().hash(state);
+    }
+}
+
+impl PartialOrd for SerializableAvroSchema {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        self.0
+            .canonical_form()
+            .partial_cmp(&other.0.canonical_form())
+    }
+}
+
+impl From<SerializableAvroSchema> for apache_avro::Schema {
+    fn from(value: SerializableAvroSchema) -> Self {
+        value.0
+    }
+}
+
+impl<'a> From<&'a SerializableAvroSchema> for &'a apache_avro::Schema {
+    fn from(value: &'a SerializableAvroSchema) -> Self {
+        &value.0
+    }
+}
+
+impl Serialize for SerializableAvroSchema {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.0.canonical_form().serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for SerializableAvroSchema {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Ok(Self(
+            apache_avro::Schema::parse_str(&String::deserialize(deserializer)?)
+                .map_err(|e| serde::de::Error::custom(format!("Invalid avro schema: {:?}", e)))?,
+        ))
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, Hash, PartialOrd, ToSchema)]
 #[serde(rename_all = "camelCase")]
-pub struct AvroFormat {}
+pub struct AvroFormat {
+    #[serde(default)]
+    pub confluent_schema_registry: bool,
+
+    #[serde(default)]
+    pub embedded_schema: bool,
+
+    #[serde(default)]
+    pub into_unstructured_json: bool,
+
+    #[serde(default)]
+    #[schema(read_only, value_type = String)]
+    pub reader_schema: Option<SerializableAvroSchema>,
+}
+
+impl AvroFormat {
+    pub fn new(
+        confluent_schema_registry: bool,
+        embedded_schema: bool,
+        into_unstructured_json: bool,
+    ) -> Self {
+        Self {
+            confluent_schema_registry,
+            embedded_schema,
+            into_unstructured_json,
+            reader_schema: None,
+        }
+    }
+
+    pub fn from_opts(opts: &mut HashMap<String, String>) -> Result<Self, String> {
+        Ok(Self::new(
+            opts.remove("avro.confluent_schema_registry")
+                .filter(|t| t == "true")
+                .is_some(),
+            opts.remove("avro.include_schema")
+                .filter(|t| t == "true")
+                .is_some(),
+            opts.remove("avro.into_unstructured_json")
+                .filter(|t| t == "true")
+                .is_some(),
+        ))
+    }
+
+    pub fn add_reader_schema(&mut self, schema: apache_avro::Schema) {
+        self.reader_schema = Some(SerializableAvroSchema(schema));
+    }
+}
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, Hash, PartialOrd, ToSchema)]
 #[serde(rename_all = "camelCase")]
@@ -118,7 +223,7 @@ impl Format {
             "json" => Format::Json(JsonFormat::from_opts(false, opts)?),
             "debezium_json" => Format::Json(JsonFormat::from_opts(true, opts)?),
             "protobuf" => return Err("protobuf is not yet supported".to_string()),
-            "avro" => return Err("avro is not yet supported".to_string()),
+            "avro" => Format::Avro(AvroFormat::from_opts(opts)?),
             "raw_string" => Format::RawString(RawStringFormat {}),
             "parquet" => Format::Parquet(ParquetFormat {}),
             f => return Err(format!("Unknown format '{}'", f)),
diff --git a/arroyo-rpc/src/lib.rs b/arroyo-rpc/src/lib.rs
index ca4431030..5b744abc5 100644
--- a/arroyo-rpc/src/lib.rs
+++ b/arroyo-rpc/src/lib.rs
@@ -1,6 +1,7 @@
 pub mod api_types;
 pub mod formats;
 pub mod public_ids;
+pub mod schema_resolver;
 
 use std::{fs, time::SystemTime};
 
diff --git a/arroyo-rpc/src/schema_resolver.rs b/arroyo-rpc/src/schema_resolver.rs
new file mode 100644
index 000000000..5309e8a90
--- /dev/null
+++ b/arroyo-rpc/src/schema_resolver.rs
@@ -0,0 +1,167 @@
+use anyhow::{anyhow, bail};
+use apache_avro::Schema;
+use async_trait::async_trait;
+use reqwest::{Client, StatusCode, Url};
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+use tracing::warn;
+
+#[async_trait]
+pub trait SchemaResolver: Send {
+    async fn resolve_schema(&self, id: u32) -> Result<Option<String>, String>;
+}
+
+/// A schema resolver that return errors when schemas are requested; this is intended
+/// to be used when schemas are embedded into the message and we do not expect to
+/// dynamically resolve them.
+pub struct FailingSchemaResolver {}
+
+impl FailingSchemaResolver {
+    pub fn new() -> Self {
+        FailingSchemaResolver {}
+    }
+}
+
+#[async_trait]
+impl SchemaResolver for FailingSchemaResolver {
+    async fn resolve_schema(&self, id: u32) -> Result<Option<String>, String> {
+        Err(format!(
+            "Schema with id {} not available, and no schema registry configured",
+            id
+        ))
+    }
+}
+
+pub struct FixedSchemaResolver {
+    id: u32,
+    schema: String,
+}
+impl FixedSchemaResolver {
+    pub fn new(id: u32, schema: Schema) -> Self {
+        FixedSchemaResolver {
+            id,
+            schema: schema.canonical_form(),
+        }
+    }
+}
+
+#[async_trait]
+impl SchemaResolver for FixedSchemaResolver {
+    async fn resolve_schema(&self, id: u32) -> Result<Option<String>, String> {
+        if id == self.id {
+            Ok(Some(self.schema.clone()))
+        } else {
+            Err(format!("Unexpected schema id {}, expected {}", id, self.id))
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Default)]
+#[serde(rename_all = "UPPERCASE")]
+pub enum ConfluentSchemaType {
+    #[default]
+    Avro,
+    Json,
+    Protobuf,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ConfluentSchemaResponse {
+    pub id: u32,
+    pub schema: String,
+    #[serde(default)]
+    pub schema_type: ConfluentSchemaType,
+    pub subject: String,
+    pub version: u32,
+}
+
+pub struct ConfluentSchemaResolver {
+    endpoint: Url,
+    topic: String,
+    client: Client,
+}
+
+impl ConfluentSchemaResolver {
+    pub fn new(endpoint: &str, topic: &str) -> anyhow::Result<Self> {
+        let client = Client::builder()
+            .timeout(Duration::from_secs(5))
+            .build()
+            .unwrap();
+
+        let endpoint: Url = format!("{}/subjects/{}-value/versions/", endpoint, topic)
+            .as_str()
+            .try_into()
+            .map_err(|_| anyhow!("{} is not a valid url", endpoint))?;
+
+        Ok(Self {
+            client,
+            topic: topic.to_string(),
+            endpoint,
+        })
+    }
+
+    pub async fn get_schema(
+        &self,
+        version: Option<u32>,
+    ) -> anyhow::Result<ConfluentSchemaResponse> {
+        let url = self
+            .endpoint
+            .join(
+                &version
+                    .map(|v| format!("{}", v))
+                    .unwrap_or_else(|| "latest".to_string()),
+            )
+            .unwrap();
+
+        let resp = self.client.get(url.clone()).send().await.map_err(|e| {
+            warn!("Got error response from schema registry: {:?}", e);
+            match e.status() {
+                Some(StatusCode::NOT_FOUND) => {
+                    anyhow!("Could not find value schema for topic '{}'", self.topic)
+                }
+
+                Some(code) => anyhow!("Schema registry returned error: {}", code),
+                None => {
+                    warn!(
+                        "Unknown error connecting to schema registry {}: {:?}",
+                        self.endpoint, e
+                    );
+                    anyhow!(
+                        "Could not connect to Schema Registry at {}: unknown error",
+                        self.endpoint
+                    )
+                }
+            }
+        })?;
+
+        if !resp.status().is_success() {
+            bail!(
+                "Received an error status code from the provided endpoint: {} {}",
+                resp.status().as_u16(),
+                resp.bytes()
+                    .await
+                    .map(|bs| String::from_utf8_lossy(&bs).to_string())
+                    .unwrap_or_else(|_| "<failed to read body>".to_string())
+            );
+        }
+
+        resp.json().await.map_err(|e| {
+            warn!(
+                "Invalid json from schema registry: {:?} for request {:?}",
+                e, url
+            );
+            anyhow!("Schema registry response could not be deserialized: {}", e)
+        })
+    }
+}
+
+#[async_trait]
+impl SchemaResolver for ConfluentSchemaResolver {
+    async fn resolve_schema(&self, id: u32) -> Result<Option<String>, String> {
+        self.get_schema(Some(id))
+            .await
+            .map(|s| Some(s.schema))
+            .map_err(|e| e.to_string())
+    }
+}
diff --git a/arroyo-sql/Cargo.toml b/arroyo-sql/Cargo.toml
index 27e89fac0..df54a7705 100644
--- a/arroyo-sql/Cargo.toml
+++ b/arroyo-sql/Cargo.toml
@@ -31,4 +31,5 @@ tracing = "0.1.37"
 
 typify = "0.0.13"
 schemars = "0.8"
-serde_json_path = "0.6.3"
\ No newline at end of file
+serde_json_path = "0.6.3"
+apache-avro = "0.16.0"
\ No newline at end of file
diff --git a/arroyo-sql/src/avro.rs b/arroyo-sql/src/avro.rs
new file mode 100644
index 000000000..7e04fb3a4
--- /dev/null
+++ b/arroyo-sql/src/avro.rs
@@ -0,0 +1,113 @@
+use crate::types::{StructDef, StructField, TypeDef};
+use anyhow::{anyhow, bail};
+use apache_avro::Schema;
+use arrow_schema::DataType;
+use proc_macro2::Ident;
+use quote::quote;
+
+pub const ROOT_NAME: &str = "ArroyoAvroRoot";
+
+pub fn convert_avro_schema(name: &str, schema: &str) -> anyhow::Result<Vec<StructField>> {
+    let schema =
+        Schema::parse_str(schema).map_err(|e| anyhow!("avro schema is not valid: {:?}", e))?;
+
+    let (typedef, _) = to_typedef(name, &schema);
+    match typedef {
+        TypeDef::StructDef(sd, _) => Ok(sd.fields),
+        TypeDef::DataType(_, _) => {
+            bail!("top-level schema must be a record")
+        }
+    }
+}
+
+pub fn get_defs(name: &str, schema: &str) -> anyhow::Result<String> {
+    let fields = convert_avro_schema(name, schema)?;
+
+    let sd = StructDef::new(Some(ROOT_NAME.to_string()), true, fields, None);
+    let defs: Vec<_> = sd
+        .all_structs_including_named()
+        .iter()
+        .map(|p| {
+            vec![
+                syn::parse_str(&p.def(false)).unwrap(),
+                p.generate_serializer_items(),
+            ]
+        })
+        .flatten()
+        .collect();
+
+    let mod_ident: Ident = syn::parse_str(name).unwrap();
+    Ok(quote! {
+        mod #mod_ident {
+            use crate::*;
+            #(#defs)
+            *
+        }
+    }
+    .to_string())
+}
+
+fn to_typedef(source_name: &str, schema: &Schema) -> (TypeDef, Option<String>) {
+    match schema {
+        Schema::Null => (TypeDef::DataType(DataType::Null, false), None),
+        Schema::Boolean => (TypeDef::DataType(DataType::Boolean, false), None),
+        Schema::Int | Schema::TimeMillis => (TypeDef::DataType(DataType::Int32, false), None),
+        Schema::Long
+        | Schema::TimeMicros
+        | Schema::TimestampMillis
+        | Schema::LocalTimestampMillis
+        | Schema::LocalTimestampMicros => (TypeDef::DataType(DataType::Int64, false), None),
+        Schema::Float => (TypeDef::DataType(DataType::Float32, false), None),
+        Schema::Double => (TypeDef::DataType(DataType::Float64, false), None),
+        Schema::Bytes | Schema::Fixed(_) | Schema::Decimal(_) => {
+            (TypeDef::DataType(DataType::Binary, false), None)
+        }
+        Schema::String | Schema::Enum(_) | Schema::Uuid => {
+            (TypeDef::DataType(DataType::Utf8, false), None)
+        }
+        Schema::Union(union) => {
+            // currently just support unions that have [t, null] as variants, which is the
+            // avro way to represent optional fields
+
+            let (nulls, not_nulls): (Vec<_>, Vec<_>) = union
+                .variants()
+                .iter()
+                .partition(|v| matches!(v, Schema::Null));
+
+            if nulls.len() == 1 && not_nulls.len() == 1 {
+                let (dt, original) = to_typedef(source_name, not_nulls[0]);
+                (dt.to_optional(), original)
+            } else {
+                (
+                    TypeDef::DataType(DataType::Utf8, false),
+                    Some("json".to_string()),
+                )
+            }
+        }
+        Schema::Record(record) => {
+            let fields = record
+                .fields
+                .iter()
+                .map(|f| {
+                    let (ft, original) = to_typedef(source_name, &f.schema);
+                    StructField::with_rename(f.name.clone(), None, ft, None, original)
+                })
+                .collect();
+
+            (
+                TypeDef::StructDef(
+                    StructDef::for_name(
+                        Some(format!("{}::{}", source_name, record.name.name)),
+                        fields,
+                    ),
+                    false,
+                ),
+                None,
+            )
+        }
+        _ => (
+            TypeDef::DataType(DataType::Utf8, false),
+            Some("json".to_string()),
+        ),
+    }
+}
diff --git a/arroyo-sql/src/lib.rs b/arroyo-sql/src/lib.rs
index 3f536ab9f..488cebf15 100644
--- a/arroyo-sql/src/lib.rs
+++ b/arroyo-sql/src/lib.rs
@@ -8,6 +8,7 @@ use arroyo_connectors::{Connection, Connector};
 use arroyo_datastream::Program;
 use datafusion::physical_plan::functions::make_scalar_function;
 
+pub mod avro;
 pub(crate) mod code_gen;
 pub mod expressions;
 pub mod external;
@@ -637,6 +638,7 @@ pub fn get_test_expression(
             KafkaConfig {
                 authentication: arroyo_connectors::kafka::KafkaConfigAuthentication::None {},
                 bootstrap_servers: "localhost:9092".to_string().try_into().unwrap(),
+                schema_registry: None,
             },
             KafkaTable {
                 topic: "test_topic".to_string(),
diff --git a/arroyo-sql/src/tables.rs b/arroyo-sql/src/tables.rs
index 2273a505d..b481419ed 100644
--- a/arroyo-sql/src/tables.rs
+++ b/arroyo-sql/src/tables.rs
@@ -24,7 +24,7 @@ use datafusion_expr::{
 use crate::code_gen::{CodeGenerator, ValuePointerContext};
 use crate::expressions::CastExpression;
 use crate::external::SinkUpdateType;
-use crate::DEFAULT_IDLE_TIME;
+use crate::{avro, DEFAULT_IDLE_TIME};
 use crate::{
     expressions::{Column, ColumnExpression, Expression, ExpressionContext},
     external::{ProcessingMode, SqlSink, SqlSource},
@@ -89,7 +89,7 @@ fn schema_type(name: &str, schema: &ConnectionSchema) -> Option<String> {
                 Some(format!("{}::{}", name, json_schema::ROOT_NAME))
             }
             SchemaDefinition::ProtobufSchema(_) => todo!(),
-            SchemaDefinition::AvroSchema(_) => todo!(),
+            SchemaDefinition::AvroSchema(_) => Some(format!("{}::{}", name, avro::ROOT_NAME)),
             SchemaDefinition::RawSchema(_) => Some("arroyo_types::RawJson".to_string()),
         }
     })
@@ -101,7 +101,7 @@ pub fn schema_defs(name: &str, schema: &ConnectionSchema) -> Option<String> {
     match def {
         SchemaDefinition::JsonSchema(s) => Some(json_schema::get_defs(&name, &s).unwrap()),
         SchemaDefinition::ProtobufSchema(_) => todo!(),
-        SchemaDefinition::AvroSchema(_) => todo!(),
+        SchemaDefinition::AvroSchema(s) => Some(avro::get_defs(&name, &s).unwrap()),
         SchemaDefinition::RawSchema(_) => None,
     }
 }
diff --git a/arroyo-sql/src/types.rs b/arroyo-sql/src/types.rs
index 9f29a8f99..af92321e6 100644
--- a/arroyo-sql/src/types.rs
+++ b/arroyo-sql/src/types.rs
@@ -727,11 +727,11 @@ impl StructField {
                     if nullable {
                         attributes.push(quote! {
                             #[serde(default)]
-                            #[serde(with = "arroyo_worker::formats::opt_timestamp_as_millis")]
+                            #[serde(with = "arroyo_worker::formats::json::opt_timestamp_as_millis")]
                         });
                     } else {
                         attributes.push(quote! {
-                            #[serde(with = "arroyo_worker::formats::timestamp_as_millis")]
+                            #[serde(with = "arroyo_worker::formats::json::timestamp_as_millis")]
                         });
                     }
                 }
@@ -739,16 +739,28 @@ impl StructField {
                     if nullable {
                         attributes.push(quote! {
                             #[serde(default)]
-                            #[serde(with = "arroyo_worker::formats::opt_timestamp_as_rfc3339")]
+                            #[serde(with = "arroyo_worker::formats::json::opt_timestamp_as_rfc3339")]
                         });
                     } else {
                         attributes.push(quote!(
-                            #[serde(with = "arroyo_worker::formats::timestamp_as_rfc3339")]
+                            #[serde(with = "arroyo_worker::formats::json::timestamp_as_rfc3339")]
                         ));
                     }
                 }
             }
+        } else if let Some("json") = self.original_type.as_ref().map(|i| i.as_str()) {
+            if self.nullable() {
+                attributes.push(quote!(
+                    #[serde(default)]
+                    #[serde(deserialize_with = "arroyo_worker::deserialize_raw_json_opt")]
+                ))
+            } else {
+                attributes.push(quote! {
+                    #[serde(deserialize_with = "arroyo_worker::deserialize_raw_json")]
+                })
+            }
         }
+
         quote! {
             #(#attributes )*
             pub #name: #type_string
diff --git a/arroyo-types/src/lib.rs b/arroyo-types/src/lib.rs
index 3ce33feb3..88a295b90 100644
--- a/arroyo-types/src/lib.rs
+++ b/arroyo-types/src/lib.rs
@@ -298,6 +298,7 @@ pub struct Record<K: Key, T: Data> {
     pub value: T,
 }
 
+#[derive(Debug, Clone)]
 pub struct UserError {
     pub name: String,
     pub details: String,
diff --git a/arroyo-worker/Cargo.toml b/arroyo-worker/Cargo.toml
index f5f1adcb0..b469cc3ec 100644
--- a/arroyo-worker/Cargo.toml
+++ b/arroyo-worker/Cargo.toml
@@ -71,6 +71,7 @@ fluvio-future = "0.6.0"
 object_store = {workspace = true }
 reqwest = "0.11.20"
 memchr = "2.6.3"
+apache-avro = "0.16.0"
 
 [dev-dependencies]
 test-case = "3"
diff --git a/arroyo-worker/src/connectors/fluvio/source.rs b/arroyo-worker/src/connectors/fluvio/source.rs
index d88e84d76..2bf7f80c2 100644
--- a/arroyo-worker/src/connectors/fluvio/source.rs
+++ b/arroyo-worker/src/connectors/fluvio/source.rs
@@ -200,7 +200,7 @@ where
                     match message {
                         Some((_, Ok(msg))) => {
                             let timestamp = from_millis(msg.timestamp().max(0) as u64);
-                            let iter = self.deserializer.deserialize_slice(msg.value());
+                            let iter = self.deserializer.deserialize_slice(msg.value()).await;
                             for value in iter {
                                 ctx.collector.collect(Record {
                                     timestamp,
diff --git a/arroyo-worker/src/connectors/kafka/source/mod.rs b/arroyo-worker/src/connectors/kafka/source/mod.rs
index 97481753a..0de53ebcc 100644
--- a/arroyo-worker/src/connectors/kafka/source/mod.rs
+++ b/arroyo-worker/src/connectors/kafka/source/mod.rs
@@ -5,6 +5,7 @@ use crate::SourceFinishType;
 use arroyo_macro::source_fn;
 use arroyo_rpc::formats::{Format, Framing};
 use arroyo_rpc::grpc::TableDescriptor;
+use arroyo_rpc::schema_resolver::{ConfluentSchemaResolver, FailingSchemaResolver, SchemaResolver};
 use arroyo_rpc::OperatorConfig;
 use arroyo_rpc::{grpc::StopMode, ControlMessage, ControlResp};
 use arroyo_state::tables::global_keyed_map::GlobalKeyedState;
@@ -17,6 +18,7 @@ use serde::de::DeserializeOwned;
 use std::collections::HashMap;
 use std::marker::PhantomData;
 use std::num::NonZeroU32;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::select;
 use tracing::{debug, error, info, warn};
@@ -103,14 +105,25 @@ where
             client_configs.insert("isolation.level".to_string(), "read_committed".to_string());
         }
 
+        let schema_resolver: Arc<dyn SchemaResolver + Sync> =
+            if let Some(schema_registry) = &connection.schema_registry {
+                Arc::new(
+                    ConfluentSchemaResolver::new(&schema_registry.endpoint, &table.topic)
+                        .expect("failed to construct confluent schema resolver"),
+                )
+            } else {
+                Arc::new(FailingSchemaResolver::new())
+            };
+
         Self {
             topic: table.topic,
             bootstrap_servers: connection.bootstrap_servers.to_string(),
             group_id: group_id.clone(),
             offset_mode: *offset,
-            deserializer: DataDeserializer::new(
+            deserializer: DataDeserializer::with_schema_resolver(
                 config.format.expect("Format must be set for Kafka source"),
                 config.framing,
+                schema_resolver,
             ),
             client_configs,
             messages_per_second: NonZeroU32::new(
@@ -247,7 +260,7 @@ where
                                     .ok_or_else(|| UserError::new("Failed to read timestamp from Kafka record",
                                         "The message read from Kafka did not contain a message timestamp"))?;
 
-                                let iter = self.deserializer.deserialize_slice(v);
+                                let iter = self.deserializer.deserialize_slice(v).await;
 
                                 for value in iter {
                                     ctx.collector.collect(Record {
diff --git a/arroyo-worker/src/connectors/kinesis/source/mod.rs b/arroyo-worker/src/connectors/kinesis/source/mod.rs
index f6af5118b..baabc9172 100644
--- a/arroyo-worker/src/connectors/kinesis/source/mod.rs
+++ b/arroyo-worker/src/connectors/kinesis/source/mod.rs
@@ -476,7 +476,7 @@ impl<K: Data, T: SchemaData> KinesisSourceFunc<K, T> {
             let data = record.data.unwrap().into_inner();
 
             let timestamp = record.approximate_arrival_timestamp.unwrap();
-            let iter = self.deserializer.deserialize_slice(&data);
+            let iter = self.deserializer.deserialize_slice(&data).await;
             for value in iter {
                 let output_record = Record {
                     timestamp: from_nanos(timestamp.as_nanos() as u128),
diff --git a/arroyo-worker/src/connectors/polling_http.rs b/arroyo-worker/src/connectors/polling_http.rs
index 591f883b3..e8acda6fa 100644
--- a/arroyo-worker/src/connectors/polling_http.rs
+++ b/arroyo-worker/src/connectors/polling_http.rs
@@ -259,7 +259,7 @@ where
                                     }
                                 }
 
-                                let iter = self.deserializer.deserialize_slice(&buf);
+                                let iter = self.deserializer.deserialize_slice(&buf).await;
 
                                 for record in iter {
                                     match record {
diff --git a/arroyo-worker/src/connectors/sse.rs b/arroyo-worker/src/connectors/sse.rs
index 8099ac29e..1e209d05f 100644
--- a/arroyo-worker/src/connectors/sse.rs
+++ b/arroyo-worker/src/connectors/sse.rs
@@ -178,7 +178,7 @@ where
                                         }
 
                                         if events.is_empty() || events.contains(&event.event_type) {
-                                            let iter = self.deserializer.deserialize_slice(&event.data.as_bytes());
+                                            let iter = self.deserializer.deserialize_slice(&event.data.as_bytes()).await;
 
                                             for v in iter {
                                                 match v {
diff --git a/arroyo-worker/src/connectors/websocket.rs b/arroyo-worker/src/connectors/websocket.rs
index 02e30a72c..2f15b3ffa 100644
--- a/arroyo-worker/src/connectors/websocket.rs
+++ b/arroyo-worker/src/connectors/websocket.rs
@@ -149,7 +149,7 @@ where
         msg: &[u8],
         ctx: &mut Context<(), T>,
     ) -> Result<(), UserError> {
-        let iter = self.deserializer.deserialize_slice(msg);
+        let iter = self.deserializer.deserialize_slice(msg).await;
         for value in iter {
             ctx.collector
                 .collect(Record {
diff --git a/arroyo-worker/src/formats/avro.rs b/arroyo-worker/src/formats/avro.rs
new file mode 100644
index 000000000..2bcd71db0
--- /dev/null
+++ b/arroyo-worker/src/formats/avro.rs
@@ -0,0 +1,481 @@
+use apache_avro::types::{Value as AvroValue, Value};
+use apache_avro::{from_avro_datum, Reader, Schema};
+use arroyo_rpc::formats::AvroFormat;
+use arroyo_rpc::schema_resolver::SchemaResolver;
+use arroyo_types::UserError;
+use serde::de::DeserializeOwned;
+use serde_json::{json, Value as JsonValue};
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tracing::info;
+
+pub async fn deserialize_slice_avro<'a, T: DeserializeOwned>(
+    format: &AvroFormat,
+    schema_registry: Arc<Mutex<HashMap<u32, Schema>>>,
+    resolver: Arc<dyn SchemaResolver + Sync>,
+    mut msg: &'a [u8],
+) -> Result<impl Iterator<Item = Result<T, UserError>> + 'a, String> {
+    let id = if format.confluent_schema_registry {
+        let magic_byte = msg[0];
+        if magic_byte != 0 {
+            return Err(format!("data was not encoded with schema registry wire format; magic byte has unexpected value: {}", magic_byte));
+        }
+
+        let id = u32::from_be_bytes([msg[1], msg[2], msg[3], msg[4]]);
+        msg = &msg[5..];
+        id
+    } else {
+        // this should be kept in sync with the id configured when we construct the
+        // FixedSchemaResolver
+        0
+    };
+
+    let mut registry = schema_registry.lock().await;
+
+    let messages = if format.embedded_schema {
+        Reader::new(&msg[..])
+            .map_err(|e| format!("invalid Avro schema in message: {:?}", e))?
+            .collect()
+    } else {
+        let schema = if registry.contains_key(&id) {
+            registry.get(&id).unwrap()
+        } else {
+            let new_schema = resolver
+                .resolve_schema(id)
+                .await?
+                .ok_or_else(|| format!("could not resolve schema for message with id {}", id))?;
+
+            let new_schema = Schema::parse_str(&new_schema).map_err(|e| {
+                format!(
+                    "schema from Confluent Schema registry is not valid: {:?}",
+                    e
+                )
+            })?;
+
+            info!("Loaded new schema with id {} from Schema Registry", id);
+            registry.insert(id, new_schema);
+
+            registry.get(&id).unwrap()
+        };
+
+        let mut buf = &msg[..];
+        vec![from_avro_datum(
+            schema,
+            &mut buf,
+            format.reader_schema.as_ref().map(|t| t.into()),
+        )]
+    };
+
+    let into_json = format.into_unstructured_json;
+    Ok(messages.into_iter().map(move |record| {
+        let value = record.map_err(|e| {
+            UserError::new(
+                "Deserialization failed",
+                format!("Failed to deserialize from avro: {:?}", e),
+            )
+        })?;
+
+        if into_json {
+            Ok(serde_json::from_value(json!({"value": avro_to_json(value).to_string()})).unwrap())
+        } else {
+            // for now round-trip through json in order to handle unsupported avro features
+            // as that allows us to rely on raw json deserialization
+            serde_json::from_value(avro_to_json(value)).map_err(|e| {
+                UserError::new(
+                    "Deserialization failed",
+                    format!("Failed to convert avro message into struct type: {:?}", e),
+                )
+            })
+        }
+    }))
+}
+
+fn convert_float(f: f64) -> JsonValue {
+    match serde_json::Number::from_f64(f) {
+        Some(n) => JsonValue::Number(n),
+        None => JsonValue::String(
+            (if f.is_infinite() && f.is_sign_positive() {
+                "+Inf"
+            } else if f.is_infinite() {
+                "-Inf"
+            } else {
+                "NaN"
+            })
+            .to_string(),
+        ),
+    }
+}
+
+fn encode_vec(v: Vec<u8>) -> JsonValue {
+    JsonValue::String(v.into_iter().map(char::from).collect())
+}
+
+fn avro_to_json(value: AvroValue) -> JsonValue {
+    match value {
+        Value::Null => JsonValue::Null,
+        Value::Boolean(b) => JsonValue::Bool(b),
+        Value::Int(i) | Value::Date(i) | Value::TimeMillis(i) => {
+            JsonValue::Number(serde_json::Number::from(i))
+        }
+        Value::Long(i)
+        | Value::TimeMicros(i)
+        | Value::TimestampMillis(i)
+        | Value::TimestampMicros(i)
+        | Value::LocalTimestampMillis(i)
+        | Value::LocalTimestampMicros(i) => JsonValue::Number(serde_json::Number::from(i)),
+        Value::Float(f) => convert_float(f as f64),
+        Value::Double(f) => convert_float(f),
+        Value::String(s) | Value::Enum(_, s) => JsonValue::String(s),
+        // this isn't the standard Avro json encoding, which just
+        Value::Bytes(b) | Value::Fixed(_, b) => encode_vec(b),
+        Value::Union(_, b) => avro_to_json(*b),
+        Value::Array(a) => JsonValue::Array(a.into_iter().map(|v| avro_to_json(v)).collect()),
+        Value::Map(m) => {
+            JsonValue::Object(m.into_iter().map(|(k, v)| (k, avro_to_json(v))).collect())
+        }
+        Value::Record(rec) => {
+            JsonValue::Object(rec.into_iter().map(|(k, v)| (k, avro_to_json(v))).collect())
+        }
+
+        Value::Decimal(d) => {
+            let b: Vec<u8> = d.try_into().unwrap_or_else(|_| vec![]);
+            encode_vec(b)
+        }
+        Value::Duration(d) => {
+            json!({
+               "months": u32::from(d.months()),
+               "days": u32::from(d.days()),
+               "milliseconds": u32::from(d.millis())
+            })
+        }
+        Value::Uuid(u) => JsonValue::String(u.to_string()),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::formats::DataDeserializer;
+    use crate::SchemaData;
+    use apache_avro::Schema;
+    use arroyo_rpc::formats::{AvroFormat, Format};
+    use arroyo_rpc::schema_resolver::{FailingSchemaResolver, FixedSchemaResolver, SchemaResolver};
+    use arroyo_types::RawJson;
+    use async_trait::async_trait;
+    use serde_json::json;
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn test_avro_deserialization() {
+        #[derive(
+            Clone,
+            Debug,
+            bincode::Encode,
+            bincode::Decode,
+            PartialEq,
+            PartialOrd,
+            serde::Serialize,
+            serde::Deserialize,
+        )]
+        pub struct ArroyoAvroRoot {
+            pub store_id: i32,
+            pub store_order_id: i32,
+            pub coupon_code: i32,
+            #[serde(deserialize_with = "crate::deserialize_raw_json")]
+            pub date: String,
+            pub status: String,
+            #[serde(deserialize_with = "crate::deserialize_raw_json")]
+            pub order_lines: String,
+        }
+        impl SchemaData for ArroyoAvroRoot {
+            fn name() -> &'static str {
+                "ArroyoAvroRoot"
+            }
+            fn schema() -> arrow::datatypes::Schema {
+                let fields: Vec<arrow::datatypes::Field> = vec![
+                    arrow::datatypes::Field::new(
+                        "store_id",
+                        arrow::datatypes::DataType::Int32,
+                        false,
+                    ),
+                    arrow::datatypes::Field::new(
+                        "store_order_id",
+                        arrow::datatypes::DataType::Int32,
+                        false,
+                    ),
+                    arrow::datatypes::Field::new(
+                        "coupon_code",
+                        arrow::datatypes::DataType::Int32,
+                        false,
+                    ),
+                    arrow::datatypes::Field::new("date", arrow::datatypes::DataType::Utf8, false),
+                    arrow::datatypes::Field::new("status", arrow::datatypes::DataType::Utf8, false),
+                    arrow::datatypes::Field::new(
+                        "order_lines",
+                        arrow::datatypes::DataType::Utf8,
+                        false,
+                    ),
+                ];
+                arrow::datatypes::Schema::new(fields)
+            }
+            fn to_raw_string(&self) -> Option<Vec<u8>> {
+                unimplemented!("to_raw_string is not implemented for this type")
+            }
+        }
+
+        let schema = r#"
+        {
+  "connect.name": "pizza_orders.pizza_orders",
+  "fields": [
+    {
+      "name": "store_id",
+      "type": "int"
+    },
+    {
+      "name": "store_order_id",
+      "type": "int"
+    },
+    {
+      "name": "coupon_code",
+      "type": "int"
+    },
+    {
+      "name": "date",
+      "type": {
+        "connect.name": "org.apache.kafka.connect.data.Date",
+        "connect.version": 1,
+        "logicalType": "date",
+        "type": "int"
+      }
+    },
+    {
+      "name": "status",
+      "type": "string"
+    },
+    {
+      "name": "order_lines",
+      "type": {
+        "items": {
+          "connect.name": "pizza_orders.order_line",
+          "fields": [
+            {
+              "name": "product_id",
+              "type": "int"
+            },
+            {
+              "name": "category",
+              "type": "string"
+            },
+            {
+              "name": "quantity",
+              "type": "int"
+            },
+            {
+              "name": "unit_price",
+              "type": "double"
+            },
+            {
+              "name": "net_price",
+              "type": "double"
+            }
+          ],
+          "name": "order_line",
+          "type": "record"
+        },
+        "type": "array"
+      }
+    }
+  ],
+  "name": "pizza_orders",
+  "namespace": "pizza_orders",
+  "type": "record"
+}"#;
+
+        let message = [
+            0u8, 0, 0, 0, 1, 8, 200, 223, 1, 144, 31, 186, 159, 2, 16, 97, 99, 99, 101, 112, 116,
+            101, 100, 4, 156, 1, 10, 112, 105, 122, 122, 97, 4, 102, 102, 102, 102, 102, 230, 38,
+            64, 102, 102, 102, 102, 102, 230, 54, 64, 84, 14, 100, 101, 115, 115, 101, 114, 116, 2,
+            113, 61, 10, 215, 163, 112, 26, 64, 113, 61, 10, 215, 163, 112, 26, 64, 0, 10,
+        ];
+
+        let mut deserializer = DataDeserializer::with_schema_resolver(
+            Format::Avro(AvroFormat::new(true, false, false)),
+            None,
+            Arc::new(FixedSchemaResolver::new(
+                1,
+                Schema::parse_str(schema).unwrap(),
+            )),
+        );
+
+        let v: Vec<Result<ArroyoAvroRoot, _>> =
+            deserializer.deserialize_slice(&message[..]).await.collect();
+
+        let expected = ArroyoAvroRoot {
+            store_id: 4,
+            store_order_id: 14308,
+            coupon_code: 1992,
+            date: "18397".to_string(),
+            status: "accepted".to_string(),
+            order_lines: "[{\"category\":\"pizza\",\"net_price\":22.9,\"product_id\":78,\"quantity\":2,\"unit_price\":11.45},{\"category\":\"dessert\",\"net_price\":6.61,\"product_id\":42,\"quantity\":1,\"unit_price\":6.61}]".to_string()
+        };
+
+        assert_eq!(v.len(), 1);
+        let v = v.into_iter().next().unwrap().unwrap();
+        // compare field-by-field because the json encoding for order_lines may not yield the
+        // same object field order
+        assert_eq!(expected.store_id, v.store_id);
+        assert_eq!(expected.store_order_id, v.store_order_id);
+        assert_eq!(expected.coupon_code, v.coupon_code);
+        assert_eq!(expected.date, v.date);
+        assert_eq!(expected.status, v.status);
+        assert_eq!(
+            serde_json::from_str::<serde_json::Value>(&expected.order_lines).unwrap(),
+            serde_json::from_str::<serde_json::Value>(&v.order_lines).unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_backwards_compatible() {
+        #[derive(
+            Clone,
+            Debug,
+            bincode::Encode,
+            bincode::Decode,
+            PartialEq,
+            PartialOrd,
+            serde::Serialize,
+            serde::Deserialize,
+        )]
+        pub struct ArroyoAvroRoot {
+            pub name: String,
+            pub favorite_number: Option<i32>,
+            pub favorite_color: Option<String>,
+            pub new_field: String,
+        }
+        impl SchemaData for ArroyoAvroRoot {
+            fn name() -> &'static str {
+                "ArroyoAvroRoot"
+            }
+            fn schema() -> arrow::datatypes::Schema {
+                unimplemented!()
+            }
+            fn to_raw_string(&self) -> Option<Vec<u8>> {
+                unimplemented!("to_raw_string is not implemented for this type")
+            }
+        }
+
+        // test schema evolution for adding a field (new_field)
+        let writer_schema = r#"{"namespace": "example.avro",
+            "type": "record",
+            "name": "User",
+            "fields": [
+                {"name": "name", "type": "string"},
+                {"name": "favorite_number",  "type": ["int", "null"]},
+                {"name": "favorite_color", "type": ["string", "null"]}
+            ]
+        }"#;
+
+        let reader_schema = r#"{"namespace": "example.avro",
+            "type": "record",
+            "name": "User",
+            "fields": [
+            {"name": "name", "type": "string"},
+            {"name": "favorite_number",  "type": ["int", "null"]},
+            {"name": "favorite_color", "type": ["string", "null"]},
+            {"name": "new_field", "type": "string", "default": "hello!"}
+            ]
+        }"#;
+
+        let mut format = AvroFormat::new(true, false, false);
+        format.add_reader_schema(Schema::parse_str(reader_schema).unwrap());
+
+        let mut deserializer = DataDeserializer::with_schema_resolver(
+            Format::Avro(format),
+            None,
+            Arc::new(FixedSchemaResolver::new(
+                1,
+                Schema::parse_str(writer_schema).unwrap(),
+            )),
+        );
+
+        let data = [0, 0, 0, 0, 1, 12, 65, 108, 121, 115, 115, 97, 0, 128, 4, 2];
+
+        let v: Result<Vec<ArroyoAvroRoot>, _> =
+            deserializer.deserialize_slice(&data[..]).await.collect();
+
+        let v = v.unwrap();
+        println!("{:?}", v);
+    }
+
+    #[tokio::test]
+    async fn test_embedded() {
+        let data = [
+            79u8, 98, 106, 1, 4, 20, 97, 118, 114, 111, 46, 99, 111, 100, 101, 99, 8, 110, 117,
+            108, 108, 22, 97, 118, 114, 111, 46, 115, 99, 104, 101, 109, 97, 186, 3, 123, 34, 116,
+            121, 112, 101, 34, 58, 32, 34, 114, 101, 99, 111, 114, 100, 34, 44, 32, 34, 110, 97,
+            109, 101, 34, 58, 32, 34, 85, 115, 101, 114, 34, 44, 32, 34, 110, 97, 109, 101, 115,
+            112, 97, 99, 101, 34, 58, 32, 34, 101, 120, 97, 109, 112, 108, 101, 46, 97, 118, 114,
+            111, 34, 44, 32, 34, 102, 105, 101, 108, 100, 115, 34, 58, 32, 91, 123, 34, 116, 121,
+            112, 101, 34, 58, 32, 34, 115, 116, 114, 105, 110, 103, 34, 44, 32, 34, 110, 97, 109,
+            101, 34, 58, 32, 34, 110, 97, 109, 101, 34, 125, 44, 32, 123, 34, 116, 121, 112, 101,
+            34, 58, 32, 91, 34, 105, 110, 116, 34, 44, 32, 34, 110, 117, 108, 108, 34, 93, 44, 32,
+            34, 110, 97, 109, 101, 34, 58, 32, 34, 102, 97, 118, 111, 114, 105, 116, 101, 95, 110,
+            117, 109, 98, 101, 114, 34, 125, 44, 32, 123, 34, 116, 121, 112, 101, 34, 58, 32, 91,
+            34, 115, 116, 114, 105, 110, 103, 34, 44, 32, 34, 110, 117, 108, 108, 34, 93, 44, 32,
+            34, 110, 97, 109, 101, 34, 58, 32, 34, 102, 97, 118, 111, 114, 105, 116, 101, 95, 99,
+            111, 108, 111, 114, 34, 125, 93, 125, 0, 52, 104, 70, 176, 108, 101, 199, 71, 44, 76,
+            126, 49, 211, 19, 204, 87, 4, 44, 12, 65, 108, 121, 115, 115, 97, 0, 128, 4, 2, 6, 66,
+            101, 110, 0, 14, 0, 6, 114, 101, 100, 52, 104, 70, 176, 108, 101, 199, 71, 44, 76, 126,
+            49, 211, 19, 204, 87,
+        ];
+
+        let mut deserializer = DataDeserializer::with_schema_resolver(
+            Format::Avro(AvroFormat::new(false, true, true)),
+            None,
+            Arc::new(FailingSchemaResolver::new()),
+        );
+
+        let v: Result<Vec<RawJson>, _> = deserializer.deserialize_slice(&data[..]).await.collect();
+
+        let v: Vec<_> = v
+            .unwrap()
+            .into_iter()
+            .map(|t| serde_json::from_str::<serde_json::Value>(&t.value).unwrap())
+            .collect();
+
+        let expected = vec![
+            json!({ "name": "Alyssa", "favorite_number": 256, "favorite_color": null }),
+            json!({ "name": "Ben", "favorite_number": 7, "favorite_color": "red" }),
+        ];
+
+        assert_eq!(v, expected);
+    }
+
+    #[tokio::test]
+    async fn test_datum_static_schema() {
+        let data = [12, 65, 108, 121, 115, 115, 97, 0, 128, 4, 2];
+
+        let schema_str = r#"{"namespace": "example.avro",
+            "type": "record",
+            "name": "User",
+            "fields": [
+            {"name": "name", "type": "string"},
+            {"name": "favorite_number",  "type": ["int", "null"]},
+            {"name": "favorite_color", "type": ["string", "null"]}
+            ]
+        }"#;
+
+        let mut format = AvroFormat::new(false, false, true);
+        format.add_reader_schema(Schema::parse_str(&schema_str).unwrap());
+        let mut deserializer = DataDeserializer::new(Format::Avro(format), None);
+
+        let v: Result<Vec<RawJson>, _> = deserializer.deserialize_slice(&data[..]).await.collect();
+
+        let expected = json!({ "name": "Alyssa", "favorite_number": 256, "favorite_color": null });
+
+        assert_eq!(
+            serde_json::from_str::<serde_json::Value>(&v.unwrap()[0].value).unwrap(),
+            expected
+        );
+    }
+}
diff --git a/arroyo-worker/src/formats.rs b/arroyo-worker/src/formats/json.rs
similarity index 62%
rename from arroyo-worker/src/formats.rs
rename to arroyo-worker/src/formats/json.rs
index 95ffdef7f..47bcfd511 100644
--- a/arroyo-worker/src/formats.rs
+++ b/arroyo-worker/src/formats/json.rs
@@ -1,15 +1,10 @@
-use std::sync::Arc;
-use std::{collections::HashMap, marker::PhantomData};
-
 use arrow::datatypes::{Field, Fields};
-use arroyo_rpc::formats::{Format, Framing, FramingMethod, JsonFormat};
-use arroyo_types::UserError;
+use arroyo_rpc::formats::JsonFormat;
 use serde::de::DeserializeOwned;
 use serde_json::{json, Value};
+use std::collections::HashMap;
 
-use crate::SchemaData;
-
-fn deserialize_slice_json<T: DeserializeOwned>(
+pub fn deserialize_slice_json<T: DeserializeOwned>(
     format: &JsonFormat,
     msg: &[u8],
 ) -> Result<T, String> {
@@ -47,152 +42,6 @@ fn deserialize_slice_json<T: DeserializeOwned>(
     }
 }
 
-fn deserialize_raw_string<T: DeserializeOwned>(msg: &[u8]) -> Result<T, String> {
-    let json = json! {
-        { "value": String::from_utf8_lossy(msg) }
-    };
-    Ok(serde_json::from_value(json).unwrap())
-}
-
-pub struct FramingIterator<'a> {
-    framing: Option<Arc<Framing>>,
-    buf: &'a [u8],
-    offset: usize,
-}
-
-impl<'a> FramingIterator<'a> {
-    pub fn new(framing: Option<Arc<Framing>>, buf: &'a [u8]) -> Self {
-        Self {
-            framing,
-            buf,
-            offset: 0,
-        }
-    }
-}
-
-impl<'a> Iterator for FramingIterator<'a> {
-    type Item = &'a [u8];
-
-    fn next(&mut self) -> Option<Self::Item> {
-        if self.offset >= self.buf.len() {
-            return None;
-        }
-
-        match &self.framing {
-            Some(framing) => {
-                match &framing.method {
-                    FramingMethod::Newline(newline) => {
-                        let end = memchr::memchr('\n' as u8, &self.buf[self.offset..])
-                            .map(|i| self.offset + i)
-                            .unwrap_or(self.buf.len());
-
-                        let prev = self.offset;
-                        self.offset = end + 1;
-
-                        // enforce max len if set
-                        let length =
-                            (end - prev).min(newline.max_line_length.unwrap_or(u64::MAX) as usize);
-
-                        Some(&self.buf[prev..(prev + length)])
-                    }
-                }
-            }
-            None => {
-                self.offset = self.buf.len();
-                Some(self.buf)
-            }
-        }
-    }
-}
-
-pub struct DataDeserializer<T: SchemaData> {
-    format: Arc<Format>,
-    framing: Option<Arc<Framing>>,
-    _t: PhantomData<T>,
-}
-
-impl<T: SchemaData> DataDeserializer<T> {
-    pub fn new(format: Format, framing: Option<Framing>) -> Self {
-        Self {
-            format: Arc::new(format),
-            framing: framing.map(|f| Arc::new(f)),
-            _t: PhantomData,
-        }
-    }
-    pub fn deserialize_slice<'a>(
-        &self,
-        msg: &'a [u8],
-    ) -> impl Iterator<Item = Result<T, UserError>> + 'a {
-        let format = self.format.clone();
-        FramingIterator::new(self.framing.clone(), msg)
-            .map(move |t| Self::deserialize_single(format.clone(), t))
-    }
-
-    fn deserialize_single(format: Arc<Format>, msg: &[u8]) -> Result<T, UserError> {
-        match &*format {
-            Format::Json(json) => deserialize_slice_json(json, msg),
-            Format::Avro(_) => todo!(),
-            Format::Parquet(_) => todo!(),
-            Format::RawString(_) => deserialize_raw_string(msg),
-        }
-        .map_err(|e| {
-            UserError::new(
-                "Deserialization failed",
-                format!(
-                    "Failed to deserialize: '{}': {}",
-                    String::from_utf8_lossy(&msg),
-                    e
-                ),
-            )
-        })
-    }
-}
-
-pub struct DataSerializer<T: SchemaData> {
-    kafka_schema: Value,
-    #[allow(unused)]
-    json_schema: Value,
-    format: Format,
-    _t: PhantomData<T>,
-}
-
-impl<T: SchemaData> DataSerializer<T> {
-    pub fn new(format: Format) -> Self {
-        Self {
-            kafka_schema: arrow_to_kafka_json(T::name(), T::schema().fields()),
-            json_schema: arrow_to_json_schema(T::schema().fields()),
-            format,
-            _t: PhantomData,
-        }
-    }
-
-    pub fn to_vec(&self, record: &T) -> Option<Vec<u8>> {
-        match &self.format {
-            Format::Json(json) => {
-                let v = if json.include_schema {
-                    let record = json! {{
-                        "schema": self.kafka_schema,
-                        "payload": record
-                    }};
-
-                    serde_json::to_vec(&record).unwrap()
-                } else {
-                    serde_json::to_vec(record).unwrap()
-                };
-
-                if json.confluent_schema_registry {
-                    todo!("Serializing to confluent schema registry is not yet supported");
-                }
-
-                Some(v)
-            }
-            Format::Avro(_) => todo!(),
-            Format::Parquet(_) => todo!(),
-            Format::RawString(_) => record.to_raw_string(),
-        }
-    }
-}
-
 #[derive(Debug)]
 pub struct MilliSecondsSystemTimeVisitor;
 
@@ -327,7 +176,10 @@ pub mod timestamp_as_rfc3339 {
         D: Deserializer<'de>,
     {
         let raw: chrono::DateTime<Utc> = DateTime::deserialize(deserializer)?;
-        Ok(from_nanos(raw.timestamp_nanos() as u128))
+        Ok(from_nanos(
+            raw.timestamp_nanos_opt()
+                .expect("could not represent time as a number of nanoseconds") as u128,
+        ))
     }
 }
 
@@ -355,11 +207,17 @@ pub mod opt_timestamp_as_rfc3339 {
         D: Deserializer<'de>,
     {
         let raw = Option::<DateTime<Utc>>::deserialize(deserializer)?;
-        Ok(raw.map(|raw| from_nanos(raw.timestamp_nanos() as u128)))
+        Ok(raw.map(|raw| {
+            from_nanos(
+                raw.timestamp_nanos_opt()
+                    .expect("could not represent time as a number of nanoseconds")
+                    as u128,
+            )
+        }))
     }
 }
 
-fn field_to_json_schema(field: &Field) -> Value {
+pub fn field_to_json_schema(field: &Field) -> Value {
     match field.data_type() {
         arrow::datatypes::DataType::Null => {
             json! {{ "type": "null" }}
@@ -417,7 +275,7 @@ fn field_to_json_schema(field: &Field) -> Value {
     }
 }
 
-fn arrow_to_json_schema(fields: &Fields) -> Value {
+pub fn arrow_to_json_schema(fields: &Fields) -> Value {
     let props: HashMap<String, Value> = fields
         .iter()
         .map(|f| (f.name().clone(), field_to_json_schema(f)))
@@ -435,7 +293,7 @@ fn arrow_to_json_schema(fields: &Fields) -> Value {
     }}
 }
 
-fn field_to_kafka_json(field: &Field) -> Value {
+pub fn field_to_kafka_json(field: &Field) -> Value {
     use arrow::datatypes::DataType::*;
 
     let typ = match field.data_type() {
@@ -503,7 +361,7 @@ fn field_to_kafka_json(field: &Field) -> Value {
 
 // For some reason Kafka uses it's own bespoke almost-but-not-quite JSON schema format
 // https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/#json-schemas
-fn arrow_to_kafka_json(name: &str, fields: &Fields) -> Value {
+pub fn arrow_to_kafka_json(name: &str, fields: &Fields) -> Value {
     let fields: Vec<_> = fields.iter().map(|f| field_to_kafka_json(&*f)).collect();
     json! {{
         "type": "struct",
@@ -512,80 +370,3 @@ fn arrow_to_kafka_json(name: &str, fields: &Fields) -> Value {
         "optional": false,
     }}
 }
-
-#[cfg(test)]
-mod tests {
-    use crate::formats::FramingIterator;
-    use arroyo_rpc::formats::{Framing, FramingMethod, NewlineDelimitedFraming};
-    use std::sync::Arc;
-
-    #[test]
-    fn test_line_framing() {
-        let framing = Some(Arc::new(Framing {
-            method: FramingMethod::Newline(NewlineDelimitedFraming {
-                max_line_length: None,
-            }),
-        }));
-
-        let result: Vec<_> = FramingIterator::new(framing.clone(), "one block".as_bytes())
-            .map(|t| String::from_utf8(t.to_vec()).unwrap())
-            .collect();
-
-        assert_eq!(vec!["one block".to_string()], result);
-
-        let result: Vec<_> = FramingIterator::new(
-            framing.clone(),
-            "one block\ntwo block\nthree block".as_bytes(),
-        )
-        .map(|t| String::from_utf8(t.to_vec()).unwrap())
-        .collect();
-
-        assert_eq!(
-            vec![
-                "one block".to_string(),
-                "two block".to_string(),
-                "three block".to_string()
-            ],
-            result
-        );
-
-        let result: Vec<_> = FramingIterator::new(
-            framing.clone(),
-            "one block\ntwo block\nthree block\n".as_bytes(),
-        )
-        .map(|t| String::from_utf8(t.to_vec()).unwrap())
-        .collect();
-
-        assert_eq!(
-            vec![
-                "one block".to_string(),
-                "two block".to_string(),
-                "three block".to_string()
-            ],
-            result
-        );
-    }
-
-    #[test]
-    fn test_max_line_length() {
-        let framing = Some(Arc::new(Framing {
-            method: FramingMethod::Newline(NewlineDelimitedFraming {
-                max_line_length: Some(5),
-            }),
-        }));
-
-        let result: Vec<_> =
-            FramingIterator::new(framing, "one block\ntwo block\nwhole".as_bytes())
-                .map(|t| String::from_utf8(t.to_vec()).unwrap())
-                .collect();
-
-        assert_eq!(
-            vec![
-                "one b".to_string(),
-                "two b".to_string(),
-                "whole".to_string()
-            ],
-            result
-        );
-    }
-}
diff --git a/arroyo-worker/src/formats/mod.rs b/arroyo-worker/src/formats/mod.rs
new file mode 100644
index 000000000..b0960bca3
--- /dev/null
+++ b/arroyo-worker/src/formats/mod.rs
@@ -0,0 +1,282 @@
+mod avro;
+pub mod json;
+
+use apache_avro::Schema;
+use std::sync::Arc;
+use std::{collections::HashMap, marker::PhantomData};
+
+use arroyo_rpc::formats::{AvroFormat, Format, Framing, FramingMethod};
+use arroyo_rpc::schema_resolver::{FailingSchemaResolver, FixedSchemaResolver, SchemaResolver};
+use arroyo_types::UserError;
+use serde::de::DeserializeOwned;
+use serde_json::{json, Value};
+use tokio::sync::Mutex;
+
+use crate::SchemaData;
+
+fn deserialize_raw_string<T: DeserializeOwned>(msg: &[u8]) -> Result<T, String> {
+    let json = json! {
+        { "value": String::from_utf8_lossy(msg) }
+    };
+    Ok(serde_json::from_value(json).unwrap())
+}
+
+pub struct FramingIterator<'a> {
+    framing: Option<Arc<Framing>>,
+    buf: &'a [u8],
+    offset: usize,
+}
+
+impl<'a> FramingIterator<'a> {
+    pub fn new(framing: Option<Arc<Framing>>, buf: &'a [u8]) -> Self {
+        Self {
+            framing,
+            buf,
+            offset: 0,
+        }
+    }
+}
+
+impl<'a> Iterator for FramingIterator<'a> {
+    type Item = &'a [u8];
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.offset >= self.buf.len() {
+            return None;
+        }
+
+        match &self.framing {
+            Some(framing) => {
+                match &framing.method {
+                    FramingMethod::Newline(newline) => {
+                        let end = memchr::memchr('\n' as u8, &self.buf[self.offset..])
+                            .map(|i| self.offset + i)
+                            .unwrap_or(self.buf.len());
+
+                        let prev = self.offset;
+                        self.offset = end + 1;
+
+                        // enforce max len if set
+                        let length =
+                            (end - prev).min(newline.max_line_length.unwrap_or(u64::MAX) as usize);
+
+                        Some(&self.buf[prev..(prev + length)])
+                    }
+                }
+            }
+            None => {
+                self.offset = self.buf.len();
+                Some(self.buf)
+            }
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct DataDeserializer<T: SchemaData> {
+    format: Arc<Format>,
+    framing: Option<Arc<Framing>>,
+    schema_registry: Arc<Mutex<HashMap<u32, Schema>>>,
+    schema_resolver: Arc<dyn SchemaResolver + Sync>,
+    _t: PhantomData<T>,
+}
+
+impl<T: SchemaData> DataDeserializer<T> {
+    pub fn new(format: Format, framing: Option<Framing>) -> Self {
+        let resolver = if let Format::Avro(AvroFormat {
+            reader_schema: Some(schema),
+            ..
+        }) = &format
+        {
+            Arc::new(FixedSchemaResolver::new(0, schema.clone().into()))
+                as Arc<dyn SchemaResolver + Sync>
+        } else {
+            Arc::new(FailingSchemaResolver::new()) as Arc<dyn SchemaResolver + Sync>
+        };
+
+        Self::with_schema_resolver(format, framing, resolver)
+    }
+
+    pub fn with_schema_resolver(
+        format: Format,
+        framing: Option<Framing>,
+        schema_resolver: Arc<dyn SchemaResolver + Sync>,
+    ) -> Self {
+        Self {
+            format: Arc::new(format),
+            framing: framing.map(|f| Arc::new(f)),
+            schema_registry: Arc::new(Mutex::new(HashMap::new())),
+            schema_resolver,
+            _t: PhantomData,
+        }
+    }
+
+    pub async fn deserialize_slice<'a>(
+        &mut self,
+        msg: &'a [u8],
+    ) -> impl Iterator<Item = Result<T, UserError>> + 'a + Send {
+        match &*self.format {
+            Format::Avro(avro) => {
+                let schema_registry = self.schema_registry.clone();
+                let schema_resolver = self.schema_resolver.clone();
+                match avro::deserialize_slice_avro(avro, schema_registry, schema_resolver, msg)
+                    .await
+                {
+                    Ok(iter) => Box::new(iter),
+                    Err(e) => Box::new(
+                        vec![Err(UserError::new("Avro deserialization failed", e))].into_iter(),
+                    )
+                        as Box<dyn Iterator<Item = Result<T, UserError>> + Send>,
+                }
+            }
+            _ => {
+                let new_self = self.clone();
+                Box::new(
+                    FramingIterator::new(self.framing.clone(), msg)
+                        .map(move |t| new_self.deserialize_single(t)),
+                ) as Box<dyn Iterator<Item = Result<T, UserError>> + Send>
+            }
+        }
+    }
+
+    fn deserialize_single(&self, msg: &[u8]) -> Result<T, UserError> {
+        match &*self.format {
+            Format::Json(json) => json::deserialize_slice_json(json, msg),
+            Format::Avro(_) => unreachable!("avro should be handled by here"),
+            Format::Parquet(_) => todo!("parquet is not supported as an input format"),
+            Format::RawString(_) => deserialize_raw_string(msg),
+        }
+        .map_err(|e| {
+            UserError::new(
+                "Deserialization failed",
+                format!(
+                    "Failed to deserialize: '{}': {}",
+                    String::from_utf8_lossy(&msg),
+                    e
+                ),
+            )
+        })
+    }
+}
+
+pub struct DataSerializer<T: SchemaData> {
+    kafka_schema: Value,
+    #[allow(unused)]
+    json_schema: Value,
+    format: Format,
+    _t: PhantomData<T>,
+}
+
+impl<T: SchemaData> DataSerializer<T> {
+    pub fn new(format: Format) -> Self {
+        Self {
+            kafka_schema: json::arrow_to_kafka_json(T::name(), T::schema().fields()),
+            json_schema: json::arrow_to_json_schema(T::schema().fields()),
+            format,
+            _t: PhantomData,
+        }
+    }
+
+    pub fn to_vec(&self, record: &T) -> Option<Vec<u8>> {
+        match &self.format {
+            Format::Json(json) => {
+                let v = if json.include_schema {
+                    let record = json! {{
+                        "schema": self.kafka_schema,
+                        "payload": record
+                    }};
+
+                    serde_json::to_vec(&record).unwrap()
+                } else {
+                    serde_json::to_vec(record).unwrap()
+                };
+
+                if json.confluent_schema_registry {
+                    todo!("Serializing to confluent schema registry is not yet supported");
+                }
+
+                Some(v)
+            }
+            Format::Avro(_) => todo!(),
+            Format::Parquet(_) => todo!(),
+            Format::RawString(_) => record.to_raw_string(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::formats::FramingIterator;
+    use arroyo_rpc::formats::{Framing, FramingMethod, NewlineDelimitedFraming};
+    use std::sync::Arc;
+
+    #[test]
+    fn test_line_framing() {
+        let framing = Some(Arc::new(Framing {
+            method: FramingMethod::Newline(NewlineDelimitedFraming {
+                max_line_length: None,
+            }),
+        }));
+
+        let result: Vec<_> = FramingIterator::new(framing.clone(), "one block".as_bytes())
+            .map(|t| String::from_utf8(t.to_vec()).unwrap())
+            .collect();
+
+        assert_eq!(vec!["one block".to_string()], result);
+
+        let result: Vec<_> = FramingIterator::new(
+            framing.clone(),
+            "one block\ntwo block\nthree block".as_bytes(),
+        )
+        .map(|t| String::from_utf8(t.to_vec()).unwrap())
+        .collect();
+
+        assert_eq!(
+            vec![
+                "one block".to_string(),
+                "two block".to_string(),
+                "three block".to_string()
+            ],
+            result
+        );
+
+        let result: Vec<_> = FramingIterator::new(
+            framing.clone(),
+            "one block\ntwo block\nthree block\n".as_bytes(),
+        )
+        .map(|t| String::from_utf8(t.to_vec()).unwrap())
+        .collect();
+
+        assert_eq!(
+            vec![
+                "one block".to_string(),
+                "two block".to_string(),
+                "three block".to_string()
+            ],
+            result
+        );
+    }
+
+    #[test]
+    fn test_max_line_length() {
+        let framing = Some(Arc::new(Framing {
+            method: FramingMethod::Newline(NewlineDelimitedFraming {
+                max_line_length: Some(5),
+            }),
+        }));
+
+        let result: Vec<_> =
+            FramingIterator::new(framing, "one block\ntwo block\nwhole".as_bytes())
+                .map(|t| String::from_utf8(t.to_vec()).unwrap())
+                .collect();
+
+        assert_eq!(
+            vec![
+                "one b".to_string(),
+                "two b".to_string(),
+                "whole".to_string()
+            ],
+            result
+        );
+    }
+}
diff --git a/connector-schemas/kafka/connection.json b/connector-schemas/kafka/connection.json
index edf3957ae..96405d6ad 100644
--- a/connector-schemas/kafka/connection.json
+++ b/connector-schemas/kafka/connection.json
@@ -49,6 +49,22 @@
                     "additionalProperties": false
                 }
             ]
+        },
+        "schemaRegistry": {
+            "type": "object",
+            "title": "Schema Registry",
+            "properties": {
+                "endpoint": {
+                    "title": "Endpoint",
+                    "type": "string",
+                    "description": "The endpoint for your Confluent Schema Registry if you have one",
+                    "examples": [
+                        "http://localhost:8081"
+                    ],
+                    "format": "uri"
+                }
+            },
+            "required": ["endpoint"]
         }
     },
     "required": [