feldera
diff --git a/‎crates/adapters/src/controller/mod.rs‎
Lines changed: 8 additions & 3 deletions b/‎crates/adapters/src/controller/mod.rs‎
Lines changed: 8 additions & 3 deletions
diff --git a/‎crates/adapters/src/format/avro/input.rs‎
Lines changed: 16 additions & 0 deletions b/‎crates/adapters/src/format/avro/input.rs‎
Lines changed: 16 additions & 0 deletions
@@ -1459,7 +1459,12 @@ impl ControllerInner {
             let format = <dyn OutputFormat>::get_format(&format_config.name).ok_or_else(|| {
                 ControllerError::unknown_output_format(endpoint_name, &format_config.name)
             })?;
-            format.new_encoder(endpoint_name, &format_config.config, &handles.schema, probe)?
+            format.new_encoder(
+                endpoint_name,
+                &endpoint_config.connector_config,
+                &handles.schema,
+                probe,
+            )?
         } else {
             // `endpoint` is `None` - instantiate an integrated endpoint.
             let endpoint = create_integrated_output_endpoint(
@@ -1940,8 +1945,8 @@ impl OutputConsumer for OutputProbe {
         }
     }
 
-    fn push_key(&mut self, key: &[u8], val: &[u8], num_records: usize) {
-        let num_bytes = key.len() + val.len();
+    fn push_key(&mut self, key: &[u8], val: Option<&[u8]>, num_records: usize) {
+        let num_bytes = key.len() + val.map(|v| v.len()).unwrap_or_default();
 
         match self.endpoint.push_key(key, val) {
             Ok(()) => {
 
@@ -140,6 +140,16 @@ impl AvroParser {
             ));
         }
 
+        match config.update_format {
+            AvroUpdateFormat::ConfluentJdbc => {
+                return Err(ControllerError::invalid_parser_configuration(
+                    endpoint_name,
+                    "'confluent_jdbc' data change event format is not yet supported by the Avro parser",
+                ));
+            }
+            AvroUpdateFormat::Debezium | AvroUpdateFormat::Raw => (),
+        }
+
         let mut parser = Self {
             endpoint_name: endpoint_name.to_string(),
             input_handle,
@@ -230,6 +240,9 @@ impl AvroParser {
                     schema_json(schema)
                 ))),
             },
+            AvroUpdateFormat::ConfluentJdbc => {
+                unreachable!()
+            }
         }
     }
 
@@ -357,6 +370,9 @@ impl AvroParser {
                         })?;
                 }
             }
+            AvroUpdateFormat::ConfluentJdbc => {
+                unreachable!()
+            }
         }
 
         Ok(())
Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,16 @@ impl AvroParser {`
`140`	`140`	`));`
`141`	`141`	`}`
`142`	`142`
	`143`	`+ match config.update_format {`
	`144`	`+ AvroUpdateFormat::ConfluentJdbc => {`
	`145`	`+ return Err(ControllerError::invalid_parser_configuration(`
	`146`	`+ endpoint_name,`
	`147`	`+ "'confluent_jdbc' data change event format is not yet supported by the Avro parser",`
	`148`	`+ ));`
	`149`	`+ }`
	`150`	`+ AvroUpdateFormat::Debezium \| AvroUpdateFormat::Raw => (),`
	`151`	`+ }`
	`152`	`+`
`143`	`153`	`let mut parser = Self {`
`144`	`154`	`endpoint_name: endpoint_name.to_string(),`
`145`	`155`	`input_handle,`
`@@ -230,6 +240,9 @@ impl AvroParser {`
`230`	`240`	`schema_json(schema)`
`231`	`241`	`))),`
`232`	`242`	`},`
	`243`	`+ AvroUpdateFormat::ConfluentJdbc => {`
	`244`	`+ unreachable!()`
	`245`	`+ }`
`233`	`246`	`}`
`234`	`247`	`}`
`235`	`248`
`@@ -357,6 +370,9 @@ impl AvroParser {`
`357`	`370`	`})?;`
`358`	`371`	`}`
`359`	`372`	`}`
	`373`	`+ AvroUpdateFormat::ConfluentJdbc => {`
	`374`	`+ unreachable!()`
	`375`	`+ }`
`360`	`376`	`}`
`361`	`377`
`362`	`378`	`Ok(())`