diff --git a/Cargo.lock b/Cargo.lock
index f9e4245e8cf96..f23640d66097a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1011,27 +1011,37 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "3.2.20"
+version = "3.2.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "23b71c3ce99b7611011217b366d923f1d0a7e07a92bb2dbf1e84508c673ca3bd"
+checksum = "86447ad904c7fb335a790c9d7fe3d0d971dc523b8ccd1561a520de9a85302750"
+dependencies = [
+ "bitflags",
+ "clap_lex 0.2.2",
+ "indexmap",
+ "textwrap",
+]
+
+[[package]]
+name = "clap"
+version = "4.0.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5840cd9093aabeabf7fd932754c435b7674520fc3ddc935c397837050f0f1e4b"
 dependencies = [
  "atty",
  "bitflags",
  "clap_derive",
- "clap_lex",
- "indexmap",
+ "clap_lex 0.3.0",
  "once_cell",
  "strsim",
  "termcolor",
- "terminal_size",
- "textwrap",
+ "terminal_size 0.2.1",
 ]
 
 [[package]]
 name = "clap_derive"
-version = "3.2.18"
+version = "4.0.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ea0c8bce528c4be4da13ea6fead8965e95b6073585a2f05204bd8f4119f82a65"
+checksum = "92289ffc6fb4a85d85c246ddb874c05a87a2e540fb6ad52f7ca07c8c1e1840b1"
 dependencies = [
  "heck",
  "proc-macro-error",
@@ -1049,6 +1059,15 @@ dependencies = [
  "os_str_bytes",
 ]
 
+[[package]]
+name = "clap_lex"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d4198f73e42b4936b35b5bb248d81d2b595ecb170da0bac7655c54eedfa8da8"
+dependencies = [
+ "os_str_bytes",
+]
+
 [[package]]
 name = "cmake"
 version = "0.1.48"
@@ -1115,7 +1134,7 @@ dependencies = [
  "encode_unicode",
  "libc",
  "once_cell",
- "terminal_size",
+ "terminal_size 0.1.17",
  "unicode-width",
  "winapi",
 ]
@@ -1244,7 +1263,7 @@ dependencies = [
  "atty",
  "cast",
  "ciborium",
- "clap",
+ "clap 3.2.22",
  "criterion-plot",
  "futures",
  "itertools",
@@ -1778,6 +1797,27 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "errno"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f639046355ee4f37944e44f60642c6f3a7efa3cf6b78c78a0d989a8ce6c396a1"
+dependencies = [
+ "errno-dragonfly",
+ "libc",
+ "winapi",
+]
+
+[[package]]
+name = "errno-dragonfly"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aa68f1b12764fab894d2755d2518754e71b4fd80ecfb822714a1206c2aab39bf"
+dependencies = [
+ "cc",
+ "libc",
+]
+
 [[package]]
 name = "ethnum"
 version = "1.3.0"
@@ -2446,6 +2486,12 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "io-lifetimes"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ea37f355c05dde75b84bba2d767906ad522e97cd9e2eef2be7a4ab7fb442c06"
+
 [[package]]
 name = "ipnet"
 version = "2.5.0"
@@ -2808,6 +2854,12 @@ dependencies = [
  "linked-hash-map",
 ]
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.0.46"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d4d2456c373231a208ad294c33dc5bff30051eafd954cd4caae83a712b12854d"
+
 [[package]]
 name = "lock_api"
 version = "0.4.6"
@@ -3018,7 +3070,7 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "axum",
- "clap",
+ "clap 4.0.8",
  "dirs",
  "indicatif",
  "mz-ore",
@@ -3138,7 +3190,7 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "chrono",
- "clap",
+ "clap 4.0.8",
  "hex",
  "mz-ore",
  "mz-test-util",
@@ -3190,7 +3242,7 @@ dependencies = [
  "anyhow",
  "axum",
  "bytesize",
- "clap",
+ "clap 4.0.8",
  "crossbeam-channel",
  "dec",
  "differential-dataflow",
@@ -3291,7 +3343,7 @@ dependencies = [
  "bytes",
  "cc",
  "chrono",
- "clap",
+ "clap 4.0.8",
  "datadriven",
  "fail",
  "fallible-iterator",
@@ -3493,7 +3545,7 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "chrono",
- "clap",
+ "clap 4.0.8",
  "crossbeam",
  "mz-avro",
  "mz-ccsr",
@@ -3599,7 +3651,7 @@ dependencies = [
  "anyhow",
  "async-trait",
  "chrono",
- "clap",
+ "clap 4.0.8",
  "futures",
  "k8s-openapi",
  "kube",
@@ -3640,7 +3692,7 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "async-trait",
- "clap",
+ "clap 4.0.8",
  "futures-core",
  "http",
  "mz-orchestrator",
@@ -3660,7 +3712,7 @@ dependencies = [
  "atty",
  "bytes",
  "chrono",
- "clap",
+ "clap 4.0.8",
  "console-subscriber",
  "ctor",
  "either",
@@ -3699,7 +3751,7 @@ dependencies = [
  "aws-sdk-kinesis",
  "aws-types",
  "bytes",
- "clap",
+ "clap 4.0.8",
  "futures",
  "mz-kinesis-util",
  "mz-ore",
@@ -3761,7 +3813,7 @@ dependencies = [
  "async-trait",
  "axum",
  "bytes",
- "clap",
+ "clap 4.0.8",
  "criterion",
  "datadriven",
  "differential-dataflow",
@@ -3830,7 +3882,7 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "bytes",
- "clap",
+ "clap 4.0.8",
  "datadriven",
  "fallible-iterator",
  "mz-ore",
@@ -4000,7 +4052,7 @@ dependencies = [
  "aws-config",
  "aws-sdk-s3",
  "bytefmt",
- "clap",
+ "clap 4.0.8",
  "futures",
  "indicatif",
  "mz-ore",
@@ -4025,7 +4077,7 @@ dependencies = [
  "anyhow",
  "async-stream",
  "async-trait",
- "clap",
+ "clap 4.0.8",
  "crossbeam-channel",
  "futures",
  "http",
@@ -4116,7 +4168,7 @@ dependencies = [
  "anyhow",
  "bytes",
  "chrono",
- "clap",
+ "clap 4.0.8",
  "fallible-iterator",
  "futures",
  "junit-report",
@@ -4250,7 +4302,7 @@ version = "0.27.0-dev"
 dependencies = [
  "anyhow",
  "axum",
- "clap",
+ "clap 4.0.8",
  "fail",
  "mz-build-info",
  "mz-http-util",
@@ -4301,7 +4353,7 @@ dependencies = [
  "byteorder",
  "bytes",
  "chrono",
- "clap",
+ "clap 4.0.8",
  "flate2",
  "futures",
  "globset",
@@ -5675,6 +5727,20 @@ dependencies = [
  "semver",
 ]
 
+[[package]]
+name = "rustix"
+version = "0.35.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72c825b8aa8010eb9ee99b75f05e10180b9278d161583034d7574c9d617aeada"
+dependencies = [
+ "bitflags",
+ "errno",
+ "io-lifetimes",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys",
+]
+
 [[package]]
 name = "ryu"
 version = "1.0.11"
@@ -6221,14 +6287,21 @@ dependencies = [
 ]
 
 [[package]]
-name = "textwrap"
-version = "0.15.0"
+name = "terminal_size"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b1141d4d61095b28419e22cb0bbf02755f5e54e0526f97f1e3d1d160e60885fb"
+checksum = "8440c860cf79def6164e4a0a983bcc2305d82419177a0e0c71930d049e3ac5a1"
 dependencies = [
- "terminal_size",
+ "rustix",
+ "windows-sys",
 ]
 
+[[package]]
+name = "textwrap"
+version = "0.15.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "949517c0cf1bf4ee812e2e07e08ab448e3ae0d23472aee8a06c985f0c8815b16"
+
 [[package]]
 name = "thiserror"
 version = "1.0.35"
diff --git a/src/billing-demo/Cargo.toml b/src/billing-demo/Cargo.toml
index 8f9662e28ba9e..48d32f6389b93 100644
--- a/src/billing-demo/Cargo.toml
+++ b/src/billing-demo/Cargo.toml
@@ -9,7 +9,7 @@ publish = false
 [dependencies]
 anyhow = "1.0.65"
 chrono = { version = "0.4.22", default-features = false, features = ["clock", "std"] }
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 hex = "0.4.3"
 mz-ore = { path = "../../src/ore", features = ["task"] }
 mz-test-util = { path = "../../test/test-util" }
diff --git a/src/compute/Cargo.toml b/src/compute/Cargo.toml
index 2a1f6539ffe51..02e79661e5364 100644
--- a/src/compute/Cargo.toml
+++ b/src/compute/Cargo.toml
@@ -10,7 +10,7 @@ publish = false
 anyhow = "1.0.65"
 axum = "0.5.16"
 bytesize = "1.1.0"
-clap = { version = "3.2.20", features = ["derive", "env"] }
+clap = { version = "4.0.8", features = ["derive", "env"] }
 crossbeam-channel = "0.5.6"
 dec = { version = "0.4.8", features = ["serde"] }
 differential-dataflow = { git = "https://github.com/TimelyDataflow/differential-dataflow.git" }
diff --git a/src/environmentd/Cargo.toml b/src/environmentd/Cargo.toml
index 7655de19b0caf..177ad0ed79ec6 100644
--- a/src/environmentd/Cargo.toml
+++ b/src/environmentd/Cargo.toml
@@ -16,7 +16,7 @@ axum = { version = "0.5.16", features = ["headers"] }
 base64 = "0.13.0"
 bytes = "1.2.1"
 chrono = { version = "0.4.22", default-features = false, features = ["std"] }
-clap = { version = "3.2.20", features = ["wrap_help", "env", "derive"] }
+clap = { version = "4.0.8", features = ["wrap_help", "env", "derive"] }
 fail = { version = "0.5.0", features = ["failpoints"] }
 futures = "0.3.24"
 headers = "0.3.7"
diff --git a/src/kafka-util/Cargo.toml b/src/kafka-util/Cargo.toml
index 1537b77965a17..58ff90c374d24 100644
--- a/src/kafka-util/Cargo.toml
+++ b/src/kafka-util/Cargo.toml
@@ -9,7 +9,7 @@ publish = false
 [dependencies]
 anyhow = "1.0.65"
 chrono = { version = "0.4.22", default-features = false, features = ["clock", "std"] }
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 crossbeam = "0.8.2"
 mz-avro = { path = "../avro" }
 mz-ccsr = { path = "../ccsr" }
diff --git a/src/mz/Cargo.toml b/src/mz/Cargo.toml
index a2a0dd0763f5a..95fd6ef3b9413 100644
--- a/src/mz/Cargo.toml
+++ b/src/mz/Cargo.toml
@@ -9,7 +9,7 @@ publish = false
 
 [dependencies]
 serde = { version = "1.0.144", features = ["derive"] }
-clap = { version = "3.2.20", features = [ "derive" ] }
+clap = { version = "4.0.8", features = [ "derive" ] }
 tokio = { version = "1.19.2", features = ["full"] }
 open = "3.0.3"
 axum = "0.5.16"
diff --git a/src/orchestrator-kubernetes/Cargo.toml b/src/orchestrator-kubernetes/Cargo.toml
index 4808969631c76..9d937081e6aff 100644
--- a/src/orchestrator-kubernetes/Cargo.toml
+++ b/src/orchestrator-kubernetes/Cargo.toml
@@ -10,7 +10,7 @@ publish = false
 anyhow = "1.0.65"
 async-trait = "0.1.57"
 chrono = { version = "0.4.22", default-features = false }
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 futures = "0.3.24"
 maplit = "1.0.2"
 mz-orchestrator = { path = "../orchestrator" }
diff --git a/src/orchestrator-tracing/Cargo.toml b/src/orchestrator-tracing/Cargo.toml
index 9a9fec3904067..bcc646131d837 100644
--- a/src/orchestrator-tracing/Cargo.toml
+++ b/src/orchestrator-tracing/Cargo.toml
@@ -9,7 +9,7 @@ publish = false
 [dependencies]
 anyhow = "1.0.65"
 async-trait = "0.1.57"
-clap = { version = "3.2.20", features = ["env", "derive"] }
+clap = { version = "4.0.8", features = ["env", "derive"] }
 futures-core = "0.3.21"
 http = "0.2.8"
 mz-orchestrator = { path = "../orchestrator" }
diff --git a/src/ore/Cargo.toml b/src/ore/Cargo.toml
index 58a6e1c13ad61..6f30e91936546 100644
--- a/src/ore/Cargo.toml
+++ b/src/ore/Cargo.toml
@@ -18,7 +18,7 @@ anyhow = { version = "1.0.65", optional = true }
 async-trait = { version = "0.1.57", optional = true }
 bytes = { version = "1.2.1", optional = true }
 chrono = { version = "0.4.22", default-features = false, features = ["std"], optional = true }
-clap = { version = "3.2.20", features = ["env"], optional = true }
+clap = { version = "4.0.8", features = ["env"], optional = true }
 ctor = { version = "0.1.23", optional = true }
 either = "1.8.0"
 futures = { version = "0.3.24", optional = true }
diff --git a/src/persist-client/Cargo.toml b/src/persist-client/Cargo.toml
index 17fe3cd8e9c02..55fc3ff378fd7 100644
--- a/src/persist-client/Cargo.toml
+++ b/src/persist-client/Cargo.toml
@@ -58,7 +58,7 @@ tokio-console = ["mz-ore/tokio-console"]
 [dev-dependencies]
 async-trait = "0.1.57"
 axum = "0.5.16"
-clap = { version = "3.2.20", features = ["derive", "env"] }
+clap = { version = "4.0.8", features = ["derive", "env"] }
 criterion = { git = "https://github.com/bheisler/criterion.rs.git", branch = "version-0.4", features = ["html_reports"] }
 datadriven = { version = "0.6.0", features = ["async"] }
 futures-task = "0.3.21"
diff --git a/src/pgtest/Cargo.toml b/src/pgtest/Cargo.toml
index 94f667a1cf26f..c9599d9ab3c93 100644
--- a/src/pgtest/Cargo.toml
+++ b/src/pgtest/Cargo.toml
@@ -9,7 +9,7 @@ publish = false
 [dependencies]
 anyhow = "1.0.65"
 bytes = "1.2.1"
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 datadriven = "0.6.0"
 fallible-iterator = "0.2.0"
 mz-ore = { path = "../ore" }
diff --git a/src/s3-datagen/Cargo.toml b/src/s3-datagen/Cargo.toml
index 9cc1beca8c18b..becc212c9ed16 100644
--- a/src/s3-datagen/Cargo.toml
+++ b/src/s3-datagen/Cargo.toml
@@ -11,7 +11,7 @@ anyhow = "1.0.65"
 aws-config = { version = "0.48.0", default-features = false, features = ["native-tls"] }
 aws-sdk-s3 = { version = "0.18.0", default-features = false, features = ["native-tls", "rt-tokio"] }
 bytefmt = "0.1.7"
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 futures = "0.3.24"
 indicatif = "0.17.1"
 mz-ore = { path = "../ore" }
diff --git a/src/service/Cargo.toml b/src/service/Cargo.toml
index 9ac9f00a25d3f..d7ab7e2f7aae2 100644
--- a/src/service/Cargo.toml
+++ b/src/service/Cargo.toml
@@ -10,7 +10,7 @@ publish = false
 anyhow = "1.0.65"
 async-stream = "0.3.3"
 async-trait = "0.1.57"
-clap = { version = "3.2.20", features = ["env", "derive"] }
+clap = { version = "4.0.8", features = ["env", "derive"] }
 crossbeam-channel = "0.5.6"
 futures = "0.3.24"
 http = "0.2.8"
diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs
index e55bbb6fcf224..2fc44221ac50e 100644
--- a/src/sql/src/plan/statement/ddl.rs
+++ b/src/sql/src/plan/statement/ddl.rs
@@ -432,7 +432,7 @@ pub fn plan_create_source(
 
                         let topic = extracted_options
                             .topic
-                            .ok_or_else(|| sql_err!("KAFKA CONNECTION without TOPIC"))?;
+                            .expect("validated exists during purification");
 
                         (
                             connection,
@@ -2765,7 +2765,8 @@ impl KafkaConnectionOptionExtracted {
                 .map_err(|e| sql_err!("parsing kafka broker: {e}"))?
                 .to_string();
             if broker.contains(',') {
-                sql_bail!("invalid CONNECTION: cannot specify multiple Kafka broker addresses in one string");
+                sql_bail!("invalid CONNECTION: cannot specify multiple Kafka broker addresses in one string.\n\n
+Instead, specify BROKERS using an array of strings, e.g. BROKERS ['kafka:9092', 'kafka:9093']");
             }
         }
 
diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs
index 2697741260a56..3f5597d865f40 100644
--- a/src/sql/src/pure.rs
+++ b/src/sql/src/pure.rs
@@ -102,7 +102,9 @@ pub async fn purify_create_source(
                     let config_options =
                         kafka_util::LibRdKafkaConfig::try_from(&extracted_options)?.0;
 
-                    let topic = extracted_options.topic.expect("validated topic exists");
+                    let topic = extracted_options
+                        .topic
+                        .ok_or_else(|| sql_err!("KAFKA CONNECTION without TOPIC"))?;
 
                     let consumer = kafka_util::create_consumer(
                         &topic,
diff --git a/src/sqllogictest/Cargo.toml b/src/sqllogictest/Cargo.toml
index 5ade3f8147bcb..7a65c4f99f277 100644
--- a/src/sqllogictest/Cargo.toml
+++ b/src/sqllogictest/Cargo.toml
@@ -10,7 +10,7 @@ publish = false
 anyhow = "1.0.65"
 bytes = "1.2.1"
 chrono = { version = "0.4.22", default-features = false, features = ["clock", "std"] }
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 fallible-iterator = "0.2.0"
 futures = "0.3.24"
 junit-report = "0.7.1"
diff --git a/src/storaged/Cargo.toml b/src/storaged/Cargo.toml
index 6ebf5ba44a7e1..409e2ac960502 100644
--- a/src/storaged/Cargo.toml
+++ b/src/storaged/Cargo.toml
@@ -9,7 +9,7 @@ publish = false
 [dependencies]
 anyhow = "1.0.65"
 axum = "0.5.16"
-clap = { version = "3.2.20", features = ["derive", "env"] }
+clap = { version = "4.0.8", features = ["derive", "env"] }
 fail = { version = "0.5.0", features = ["failpoints"] }
 mz-build-info = { path = "../build-info" }
 mz-http-util = { path = "../http-util" }
diff --git a/src/testdrive/Cargo.toml b/src/testdrive/Cargo.toml
index dba3489c9ee34..a8ac789c57330 100644
--- a/src/testdrive/Cargo.toml
+++ b/src/testdrive/Cargo.toml
@@ -21,7 +21,7 @@ aws-types = { version = "0.48.0", features = ["hardcoded-credentials"] }
 byteorder = "1.4.3"
 bytes = "1.2.1"
 chrono = { version = "0.4.22", default-features = false, features = ["std"] }
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 flate2 = "1.0.24"
 futures = "0.3.24"
 globset = "0.4.9"
diff --git a/test/perf-kinesis/Cargo.toml b/test/perf-kinesis/Cargo.toml
index 0f03125856053..4c5cd530681d1 100644
--- a/test/perf-kinesis/Cargo.toml
+++ b/test/perf-kinesis/Cargo.toml
@@ -12,7 +12,7 @@ aws-config = { version = "0.48.0", default-features = false, features = ["native
 aws-sdk-kinesis = { version = "0.18.0", default-features = false, features = ["native-tls", "rt-tokio"] }
 aws-types = "0.48.0"
 bytes = "1.2.1"
-clap = { version = "3.2.20", features = ["derive"] }
+clap = { version = "4.0.8", features = ["derive"] }
 futures = "0.3.24"
 mz-kinesis-util = { path = "../../src/kinesis-util" }
 mz-ore = { path = "../../src/ore", features = ["task"] }
diff --git a/test/testdrive/kafka-sources.td b/test/testdrive/kafka-sources.td
index d8d57b94da248..d0a2d808eadbb 100644
--- a/test/testdrive/kafka-sources.td
+++ b/test/testdrive/kafka-sources.td
@@ -12,3 +12,10 @@
 ! CREATE SOURCE s
   FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC 'testdrive-data-${testdrive.seed}'
 contains:Source format must be specified
+
+> CREATE CONNECTION kafka_conn
+  FOR KAFKA BROKER '';
+
+! CREATE SOURCE s
+  FROM KAFKA CONNECTION kafka_conn
+contains:KAFKA CONNECTION without TOPIC