async fn publish_kafka_schemas(
    ccsr: &ccsr::Client,
    topic: &str,
    key_schema: Option<&str>,
    key_schema_type: Option<ccsr::SchemaType>,
    value_schema: &str,
    value_schema_type: ccsr::SchemaType,
) -> Result<(Option<i32>, i32), CoordError> {
    let value_schema_id = ccsr
        .publish_schema(
            &format!("{}-value", topic),
            value_schema,
            value_schema_type,
            &[],
        )
        .await
        .context("unable to publish value schema to registry in kafka sink")?;

    let key_schema_id = if let Some(key_schema) = key_schema {
        let key_schema_type = key_schema_type.ok_or_else(|| {
            CoordError::Unstructured(anyhow!("expected schema type for key schema"))
        })?;
        Some(
            ccsr.publish_schema(&format!("{}-key", topic), key_schema, key_schema_type, &[])
                .await
                .context("unable to publish key schema to registry in kafka sink")?,
        )
    } else {
        None
    };

    Ok((key_schema_id, value_schema_id))
}
/// Publish value and optional key schemas for a given topic.
///
/// TODO(benesch): do we need to delete the Kafka topic if publishing the
// schema fails?
Publish value and optional key schemas for a given topic.
Publish value and optional key schemas for a given topic .
async fn publish_kafka_schemas(
    ccsr: &ccsr::Client,
    topic: &str,
    key_schema: Option<&str>,
    key_schema_type: Option<ccsr::SchemaType>,
    value_schema: &str,
    value_schema_type: ccsr::SchemaType,
) -> Result<(Option<i32>, i32), CoordError> {
    let value_schema_id = ccsr
        .publish_schema(
            &format!("{}-value", topic),
            value_schema,
            value_schema_type,
            &[],
        )
        .await
        .context("unable to publish value schema to registry in kafka sink")?;
    let key_schema_id = if let Some(key_schema) = key_schema {
        let key_schema_type = key_schema_type.ok_or_else(|| {
            CoordError::Unstructured(anyhow!("expected schema type for key schema"))
        })?;
        Some(
            ccsr.publish_schema(&format!("{}-key", topic), key_schema, key_schema_type, &[])
                .await
                .context("unable to publish key schema to registry in kafka sink")?,
        )
    } else {
        None
    };
    Ok((key_schema_id, value_schema_id))
}
async fn publish_kafka_schemas ( ccsr : & ccsr :: Client , topic : & str , key_schema : Option < & str > , key_schema_type : Option < ccsr :: SchemaType > , value_schema : & str , value_schema_type : ccsr :: SchemaType , ) -> Result < ( Option < i32 > , i32 ) , CoordError > { let value_schema_id = ccsr . publish_schema ( & format ! ( " " topic ) , value_schema , value_schema_type , & [ ] , ) . await . context ( " " ) ? ; let key_schema_id = if let Some ( key_schema ) = key_schema { let key_schema_type = key_schema_type . ok_or_else ( | | { CoordError :: Unstructured ( anyhow ! ( " " ) ) } ) ? ; Some ( ccsr . publish_schema ( & format ! ( " " topic ) , key_schema , key_schema_type , & [ ] ) . await . context ( " " ) ? , ) } else { None } ; Ok ( ( key_schema_id , value_schema_id ) ) }
Publish value and optional key schemas for a given topic.
Publish value and optional key schemas for a given topic .
ccsr
&ccsr::Client
topic
&str
key_schema
Option<&str>
key_schema_type
Option<ccsr::SchemaType>
value_schema
&str
value_schema_type
ccsr::SchemaType
ccsr
&ccsr::Client
topic
&str
key_schema
Option<&str>
key_schema_type
Option<ccsr::SchemaType>
value_schema
&str
value_schema_type
ccsr::SchemaType