Dekaf materialization endpoint support

crates/agent/src/api/authorize_task.rs

@@ -129,7 +129,6 @@ fn evaluate_authorization(
         tasks,
         ..
     } = snapshot;
-
     // Map `claims.sub`, a Shard ID, into its task.
     let task = tasks
         .binary_search_by(|task| {

crates/dekaf/Cargo.toml

@@ -21,6 +21,7 @@ models = { path = "../models" }
 ops = { path = "../ops" }
 proto-flow = { path = "../proto-flow" }
 proto-gazette = { path = "../proto-gazette" }
+runtime = { path = "../runtime" }
 simd-doc = { path = "../simd-doc" }
 
 aes-siv = { workspace = true }
@@ -39,6 +40,7 @@ hex = { workspace = true }
 hexdump = { workspace = true }
 humantime = { workspace = true }
 itertools = { workspace = true }
+jsonwebtoken = { workspace = true }
 kafka-protocol = { workspace = true }
 lazy_static = { workspace = true }
 lz4_flex = { workspace = true }
@@ -58,6 +60,7 @@ schemars = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 socket2 = { workspace = true }
+sqlx = { workspace = true }
 time = { workspace = true }
 tokio = { workspace = true }
 tokio-rustls = { workspace = true }

crates/dekaf/src/api_client.rs

@@ -1,7 +1,6 @@
 use anyhow::{anyhow, bail, Context};
 use bytes::{Bytes, BytesMut};
 use futures::{SinkExt, TryStreamExt};
-use gazette::broker;
 use kafka_protocol::{
     error::ParseResponseErrorCode,
     messages::{self, ApiKey},
@@ -11,17 +10,15 @@ use rsasl::{config::SASLConfig, mechname::Mechname, prelude::SASLClient};
 use rustls::RootCertStore;
 use std::{
     boxed::Box,
-    cell::Cell,
     collections::HashMap,
     fmt::Debug,
     io,
     time::{Duration, SystemTime},
 };
 use std::{io::BufWriter, pin::Pin, sync::Arc};
 use tokio::sync::OnceCell;
-use tokio::sync::RwLock;
 use tokio_rustls::rustls;
-use tokio_util::{codec, task::AbortOnDropHandle};
+use tokio_util::codec;
 use tracing::instrument;
 use url::Url;
 

crates/dekaf/src/connector.rs

@@ -1,5 +1,5 @@
 use anyhow::{bail, Context};
-use proto_flow::materialize;
+use proto_flow::{flow::materialization_spec, materialize};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
@@ -38,6 +38,7 @@ pub struct DekafConfig {
     /// tombstones with null values, and "Header" emits then as a kafka document
     /// with empty string and `_is_deleted` header set to `1`. Setting this value
     /// will also cause all other non-deletions to have an `_is_deleted` header of `0`.
+    #[serde(default)]
     pub deletions: DeletionMode,
 }
 
@@ -70,8 +71,22 @@ pub async fn unary_materialize(
 ) -> anyhow::Result<materialize::Response> {
     use proto_flow::materialize::response::validated;
     if let Some(mut validate) = request.validate {
-        serde_json::de::from_str::<DekafConfig>(&validate.config_json)
-            .context("validating endpoint config")?;
+        match materialization_spec::ConnectorType::try_from(validate.connector_type)? {
+            materialization_spec::ConnectorType::Dekaf => {}
+            other => bail!("invalid connector type: {}", other.as_str_name()),
+        };
+
+        let parsed_outer_config =
+            serde_json::from_str::<models::DekafConfig>(&validate.config_json)
+                .context("validating dekaf config")?;
+
+        let _parsed_inner_config = serde_json::from_value::<DekafConfig>(
+            parsed_outer_config.config.to_value(),
+        )
+        .context(format!(
+            "validating dekaf endpoint config for variant {}",
+            parsed_outer_config.variant
+        ))?;
 
         // Largely copied from crates/validation/src/noop.rs
         let validated_bindings = std::mem::take(&mut validate.bindings)