Updates

arroyo-api/src/pipelines.rs

@@ -32,7 +32,7 @@ use arroyo_formats::json::arrow_to_json_schema;
 use arroyo_rpc::formats::Format;
 use arroyo_rpc::public_ids::{generate_id, IdTypes};
 use arroyo_rpc::schema_resolver::{ConfluentSchemaRegistry, ConfluentSchemaType};
-use arroyo_rpc::OperatorConfig;
+use arroyo_rpc::{error_chain, OperatorConfig};
 use arroyo_server_common::log_event;
 use arroyo_sql::types::StructDef;
 use arroyo_sql::{has_duplicate_udf_names, ArroyoSchemaProvider, CompiledSql, SqlConfig};
@@ -194,8 +194,7 @@ async fn try_register_confluent_schema(
 
                 let id = schema_registry
                     .write_schema(schema.canonical_form(), ConfluentSchemaType::Avro)
-                    .await
-                    .map_err(|e| anyhow!("Failed to write schema to schema registry: {}", e))?;
+                    .await?;
 
                 avro.schema_id = Some(id as u32);
                 config.format = Some(Format::Avro(avro))
@@ -209,8 +208,7 @@ async fn try_register_confluent_schema(
 
                 let id = schema_registry
                     .write_schema(schema.to_string(), ConfluentSchemaType::Json)
-                    .await
-                    .map_err(|e| anyhow!("Failed to write schema to schema registry: {}", e))?;
+                    .await?;
 
                 json.schema_id = Some(id as u32);
                 config.format = Some(Format::Json(json))
@@ -346,7 +344,7 @@ pub(crate) async fn create_pipeline<'a>(
             message: format!(
                 "Failed to register schemas with the schema registry. Make sure \
             that the schema_registry is configured correctly and running.\nDetails: {}",
-                e
+                error_chain(e)
             ),
         })?;
 

arroyo-connectors/src/kafka.rs

@@ -497,7 +497,7 @@ impl KafkaTester {
                         .await
                         .map_err(|e|
                             if msg[0] == 0 {
-                                anyhow!("Failed to parse message are regular Avro. It may be encoded as SR-Avro, but the schema registry is not enabled. Ensure that the format and schema type are correct.")
+                                anyhow!("Failed to parse message as regular Avro. It may be encoded as SR-Avro, but the schema registry is not enabled. Ensure that the format and schema type are correct.")
                             } else {
                                 anyhow!("Failed to parse message as Avro: {:?}. Ensure that the format and schema type are correct.", e)
                             })?;

arroyo-console/src/components/StartPipelineModal.tsx

@@ -37,7 +37,7 @@ const StartPipelineModal: React.FC<StartPipelineModalProps> = ({
   start,
 }) => {
   return (
-    <Modal isOpen={isOpen} onClose={onClose} isCentered>
+    <Modal isOpen={isOpen} onClose={onClose} isCentered size={startError ? '4xl' : 'xl'}>
       <ModalOverlay />
       <ModalContent>
         <ModalHeader>Start Pipeline</ModalHeader>
@@ -47,7 +47,9 @@ const StartPipelineModal: React.FC<StartPipelineModalProps> = ({
             {startError ? (
               <Alert status="error">
                 <AlertIcon />
-                <AlertDescription>{startError}</AlertDescription>
+                <AlertDescription overflowY={"auto"} maxH={400} whiteSpace={'pre-wrap'}>{
+                  startError
+                }</AlertDescription>
               </Alert>
             ) : null}
 

arroyo-console/src/lib/data_fetching.ts

@@ -150,7 +150,7 @@ const pingFetcher = async () => {
 
 export const usePing = () => {
   const { data, error, isLoading } = useSWR('ping', pingFetcher, {
-    refreshInterval: 100000000,
+    refreshInterval: 1000,
     onErrorRetry: (error, key, config, revalidate, {}) => {
       // explicitly define this function to override the exponential backoff
       setTimeout(() => revalidate(), 1000);
@@ -251,7 +251,10 @@ const connectionProfileAutocompleteFetcher = () => {
 export const useConnectionProfileAutocomplete = (id: string) => {
   const { data, error } = useSWR<schemas['ConnectionAutocompleteResp']>(
     connectionProfileAutocompleteKey(id),
-    connectionProfileAutocompleteFetcher()
+    connectionProfileAutocompleteFetcher(),
+    {
+      revalidateOnMount: true,
+    }
   );
 
   return {

arroyo-console/src/routes/connections/ConfluentSchemaEditor.tsx

@@ -13,7 +13,7 @@ export function ConfluentSchemaEditor({
 }) {
   return (
     <Stack spacing={4} maxW="lg">
-      <Text>Schemas will be loaded from the configured Confluent Schema Registry</Text>
+      <Text>Schemas will be loaded from and written to the configured Confluent Schema Registry</Text>
 
       <Button colorScheme="green" onClick={() => next()}>
         Continue

arroyo-rpc/src/lib.rs

@@ -182,3 +182,11 @@ impl Default for OperatorConfig {
         }
     }
 }
+
+pub fn error_chain(e: anyhow::Error) -> String {
+    e.chain()
+        .into_iter()
+        .map(|e| e.to_string())
+        .collect::<Vec<_>>()
+        .join(": ")
+}

arroyo-rpc/src/schema_resolver.rs

@@ -231,7 +231,11 @@ impl ConfluentSchemaRegistryClient {
 
             match status {
                 StatusCode::CONFLICT => {
-                    bail!("{}", body)
+                    bail!(
+                        "there is already an existing schema for this topic which is \
+                    incompatible with the new schema being registered:\n\n{}",
+                        body
+                    )
                 }
                 StatusCode::UNPROCESSABLE_ENTITY => {
                     bail!("invalid schema: {}", body);
@@ -333,7 +337,7 @@ impl ConfluentSchemaRegistry {
         self.client
             .write_schema(self.topic_endpoint(), schema, schema_type)
             .await
-            .context(format!("failed to write schema for topic '{}'", self.topic))
+            .context(format!("topic '{}'", self.topic))
     }
 
     pub async fn get_schema_for_id(