dekaf: keep connection open on document validation error

crates/dekaf/src/lib.rs

@@ -346,7 +346,7 @@ impl App {
 /// Dispatch a read request `frame` of the current session, writing its response into `out`.
 /// `raw_sasl_auth` is the state of SASL "raw" mode authentication,
 /// and conditions the interpretation of request frames.
-#[tracing::instrument(level = "trace", err(level = "warn"), skip_all)]
+#[tracing::instrument(level = "trace", skip_all)]
 pub async fn dispatch_request_frame(
     session: &mut Session,
     raw_sasl_auth: &mut bool,

crates/dekaf/src/main.rs

@@ -115,6 +115,11 @@ pub struct Cli {
     #[arg(long, env = "SPEC_TTL", value_parser = humantime::parse_duration, default_value = "2m")]
     spec_ttl: std::time::Duration,
 
+    /// How long a connection will keep returning LeaderNotAvailable for a partition
+    /// stuck failing schema validation before giving up and closing the connection.
+    #[arg(long, env = "SCHEMA_ERROR_COOLDOWN_TIMEOUT", value_parser = humantime::parse_duration, default_value = "4m")]
+    schema_error_cooldown_timeout: std::time::Duration,
+
     /// Timeout for TLS handshake completion
     #[arg(long, env = "TLS_HANDSHAKE_TIMEOUT", value_parser = humantime::parse_duration, default_value = "10s")]
     tls_handshake_timeout: std::time::Duration,
@@ -434,6 +439,7 @@ async fn async_main(cli: Cli) -> anyhow::Result<()> {
                                 upstream_auth.clone(),
                                 cli.read_buffer_chunk_limit,
                                 cli.combined_partition_fetch_limit,
+                                cli.schema_error_cooldown_timeout,
                             ),
                             socket,
                             tls_acceptor.clone(),

crates/dekaf/src/topology.rs

@@ -58,6 +58,7 @@ pub struct Collection {
     pub spec: flow::CollectionSpec,
     pub uuid_ptr: json::Pointer,
     pub value_schema: avro::Schema,
+    pub schema_hash: String,
     pub extractors: Vec<(avro::Schema, utils::CustomizableExtractor)>,
     pub binding_backfill_counter: u32,
 }
@@ -261,6 +262,17 @@ impl Collection {
 
         let key_schema = avro::key_to_avro(&key_ptr, collection_schema_shape);
 
+        // Content-addresses the derived key/value schemas so callers can
+        // cheaply detect when a binding's effective schema has changed,
+        // without a network round trip (unlike `registered_schema_id`'s
+        // `avro_schema_md5`, which addresses the same schemas against the
+        // control plane's schema registry table).
+        let schema_hash = {
+            let key_json = serde_json::to_value(&key_schema).unwrap().to_string();
+            let value_json = serde_json::to_value(&value_schema).unwrap().to_string();
+            format!("{:x}", md5::compute(format!("{key_json}{value_json}")))
+        };
+
         let (mut not_before, not_after) = (
             binding.not_before.map(|b| {
                 uuid::Clock::from_unix(b.seconds.try_into().unwrap(), b.nanos.try_into().unwrap())
@@ -315,6 +327,7 @@ impl Collection {
             spec: collection_spec,
             uuid_ptr,
             value_schema,
+            schema_hash,
             extractors,
             // Start the backfill counter (which will map to the topic leader epoch) at 1, not 0.
             // Kafka consumers don't seem to handle going from epoch 0 to epoch 1 gracefully. Specifically,

crates/dekaf/tests/e2e/fetch_offsets.rs

@@ -1,54 +1,22 @@
 use super::DekafTestEnv;
-use super::raw_kafka::TestKafkaClient;
+use super::raw_kafka::{TestKafkaClient, decode_fetch_records};
 use anyhow::Context;
-use bytes::Buf;
-use kafka_protocol::messages;
-use kafka_protocol::records::{Record, RecordBatchDecoder};
+use kafka_protocol::records::Record;
 use serde_json::json;
 use std::time::Duration;
 
 const FIXTURE: &str = include_str!("fixtures/basic.flow.yaml");
 const TOPIC: &str = "test_topic";
 const PARTITION: i32 = 0;
 
-/// Decode all records, including control records, from a raw FetchResponse.
-///
-/// Unlike the rdkafka consumer, this surfaces exactly what Dekaf put on the
-/// wire: librdkafka silently filters out control records and records below
-/// the fetch offset, which masks bugs in which documents a fetch serves.
-fn decode_fetch_records(resp: &messages::FetchResponse) -> anyhow::Result<Vec<Record>> {
-    let partition = resp
-        .responses
-        .iter()
-        .find(|t| t.topic.as_str() == TOPIC)
-        .and_then(|t| t.partitions.iter().find(|p| p.partition_index == PARTITION))
-        .context("missing partition in fetch response")?;
-
-    anyhow::ensure!(
-        partition.error_code == 0,
-        "fetch returned error code {}",
-        partition.error_code
-    );
-
-    let Some(mut buf) = partition.records.clone() else {
-        return Ok(Vec::new());
-    };
-
-    let mut records = Vec::new();
-    while buf.has_remaining() {
-        records.extend(RecordBatchDecoder::decode(&mut buf)?.records);
-    }
-    Ok(records)
-}
-
 async fn fetch_records_at(
     client: &mut TestKafkaClient,
     offset: i64,
 ) -> anyhow::Result<Vec<Record>> {
     let resp = client
         .fetch_with_epoch(TOPIC, PARTITION, offset, -1)
         .await?;
-    decode_fetch_records(&resp)
+    decode_fetch_records(&resp, TOPIC, PARTITION)
 }
 
 /// Fetch at `offset`, retrying empty responses (e.g. while the server-side

crates/dekaf/tests/e2e/fixtures/schema_cooldown.flow.yaml

@@ -0,0 +1,57 @@
+collections:
+    test_data_a:
+        writeSchema:
+            type: object
+            properties:
+                id: { type: string }
+                value: { type: [integer, string] }
+            required: [id]
+        readSchema:
+            type: object
+            properties:
+                id: { type: string }
+                value: { type: integer }
+            required: [id]
+        key: [/id]
+
+    test_data_b:
+        schema:
+            type: object
+            properties:
+                id: { type: string }
+                value: { type: integer }
+            required: [id]
+        key: [/id]
+
+captures:
+    source_ingest:
+        endpoint:
+            connector:
+                image: ghcr.io/estuary/source-http-ingest:dev
+                config:
+                    paths: ["/data_a", "/data_b"]
+        bindings:
+            - resource: { path: "/data_a", stream: "/data_a" }
+              target: test_data_a
+            - resource: { path: "/data_b", stream: "/data_b" }
+              target: test_data_b
+
+materializations:
+    dekaf_test:
+        endpoint:
+            dekaf:
+                variant: testing
+                config:
+                    token: "test-token-12345"
+                    strict_topic_names: false
+        bindings:
+            - source: test_data_a
+              resource: { topic_name: topic_a }
+              fields:
+                recommended: true
+                exclude: [flow_published_at]
+            - source: test_data_b
+              resource: { topic_name: topic_b }
+              fields:
+                recommended: true
+                exclude: [flow_published_at]

crates/dekaf/tests/e2e/harness.rs

@@ -526,6 +526,36 @@ impl DekafTestEnv {
         self.publish_catalog(&catalog).await
     }
 
+    /// Republish `collection_name` with a new `read_schema`, leaving its
+    /// `write_schema` untouched. The fixture must already declare separate
+    /// `writeSchema`/`readSchema` (not a single `schema`) for this to apply.
+    pub async fn set_collection_read_schema(
+        &self,
+        collection_name: &str,
+        read_schema: serde_json::Value,
+    ) -> anyhow::Result<()> {
+        let collection = models::Collection::new(collection_name);
+        let mut coll_def = self
+            .catalog
+            .collections
+            .get(&collection)
+            .context("collection not in fixture")?
+            .clone();
+
+        coll_def.read_schema = Some(models::Schema::new(models::RawValue::from_value(
+            &read_schema,
+        )));
+
+        tracing::info!(%collection_name, "Updating collection read schema");
+
+        let catalog = models::Catalog {
+            collections: [(collection, coll_def)].into(),
+            ..Default::default()
+        };
+
+        self.publish_catalog(&catalog).await
+    }
+
     /// Cleanup test specs synchronously.
     fn cleanup_sync(&self) {
         let cmd = match flowctl_command() {

crates/dekaf/tests/e2e/main.rs

@@ -12,6 +12,7 @@ mod list_offsets;
 mod migration;
 mod not_ready;
 mod partition_eofs;
+mod schema_cooldown;
 
 pub use harness::{
     ConnectionInfo, DekafTestEnv, cluster_name, cluster_name_2, connection_info_for_dataplane,

crates/dekaf/tests/e2e/raw_kafka.rs

@@ -1,7 +1,10 @@
+use anyhow::Context;
+use bytes::Buf;
 use dekaf::{KafkaApiClient, KafkaClientAuth};
 use kafka_protocol::{
     messages::{self, offset_fetch_response::OffsetFetchResponsePartition},
     protocol::StrBytes,
+    records::{Record, RecordBatchDecoder},
 };
 
 /// Protocol versions to use for test requests.
@@ -66,6 +69,40 @@ impl TestKafkaClient {
         self.inner.send_request(req, Some(header)).await
     }
 
+    /// Fetch several `(topic, partition, fetch_offset)` requests in a single
+    /// call, on one connection — for asserting that a broken partition's
+    /// error doesn't affect a sibling partition's Fetch response.
+    pub async fn fetch_multi(
+        &mut self,
+        requests: &[(&str, i32, i64)],
+    ) -> anyhow::Result<messages::FetchResponse> {
+        let req = messages::FetchRequest::default()
+            .with_max_wait_ms(1000)
+            .with_min_bytes(1)
+            .with_max_bytes(1024 * 1024)
+            .with_topics(
+                requests
+                    .iter()
+                    .map(|&(topic, partition, offset)| {
+                        messages::fetch_request::FetchTopic::default()
+                            .with_topic(topic_name(topic))
+                            .with_partitions(vec![
+                                messages::fetch_request::FetchPartition::default()
+                                    .with_partition(partition)
+                                    .with_fetch_offset(offset)
+                                    .with_partition_max_bytes(1024 * 1024),
+                            ])
+                    })
+                    .collect(),
+            );
+
+        let header = messages::RequestHeader::default()
+            .with_request_api_key(messages::ApiKey::Fetch as i16)
+            .with_request_api_version(protocol_versions::FETCH);
+
+        self.inner.send_request(req, Some(header)).await
+    }
+
     /// ListOffsets with explicit `current_leader_epoch`.
     ///
     /// - `timestamp = -2`: earliest offset
@@ -195,6 +232,40 @@ impl TestKafkaClient {
     }
 }
 
+/// Decode all records, including control records, from a raw FetchResponse.
+///
+/// Unlike the rdkafka consumer, this surfaces exactly what Dekaf put on the
+/// wire: librdkafka silently filters out control records and records below
+/// the fetch offset, which masks bugs in which documents a fetch serves.
+pub fn decode_fetch_records(
+    resp: &messages::FetchResponse,
+    topic: &str,
+    partition: i32,
+) -> anyhow::Result<Vec<Record>> {
+    let partition_data = resp
+        .responses
+        .iter()
+        .find(|t| t.topic.as_str() == topic)
+        .and_then(|t| t.partitions.iter().find(|p| p.partition_index == partition))
+        .context("missing partition in fetch response")?;
+
+    anyhow::ensure!(
+        partition_data.error_code == 0,
+        "fetch returned error code {}",
+        partition_data.error_code
+    );
+
+    let Some(mut buf) = partition_data.records.clone() else {
+        return Ok(Vec::new());
+    };
+
+    let mut records = Vec::new();
+    while buf.has_remaining() {
+        records.extend(RecordBatchDecoder::decode(&mut buf)?.records);
+    }
+    Ok(records)
+}
+
 /// Extract the error code from a FetchResponse for a specific topic/partition.
 pub fn fetch_partition_error(
     resp: &messages::FetchResponse,