dekaf: Allow for indirecting DekafConfig if/when desired

crates/flowctl/src/generate/mod.rs

@@ -3,6 +3,8 @@ use anyhow::Context;
 use futures::{FutureExt, StreamExt};
 use itertools::Itertools;
 use proto_flow::{capture, derive, flow, materialize};
+use std::str::FromStr;
+use url::Url;
 
 #[derive(Debug, clap::Args)]
 #[clap(rename_all = "kebab-case")]
@@ -293,8 +295,10 @@ async fn generate_missing_materialization_configs(
  connector_type: flow::materialization_spec::ConnectorType::Dekaf as i32,
  config_json: serde_json::to_string(config).unwrap(),
  },
- // Dekaf isn't a pluggable connector, and so does not have dynamic config.
- None,
+ match &config {
+ models::DekafConfigContainer::Indirect(s) => Url::from_str(s.as_str()).ok(),
+ _ => None,
+ },
  ),
  };
  let missing_resource_urls: Vec<(url::Url, models::Collection)> = bindings

crates/models/src/lib.rs

@@ -34,8 +34,8 @@ pub use journals::{
  AZURE_STORAGE_ACCOUNT_RE, GCS_BUCKET_RE, S3_BUCKET_RE,
 };
 pub use materializations::{
- MaterializationBinding, MaterializationDef, MaterializationEndpoint, MaterializationFields,
- SqliteConfig,
+ DekafConfigContainer, MaterializationBinding, MaterializationDef, MaterializationEndpoint,
+ MaterializationFields, SqliteConfig,
 };
 pub use raw_value::RawValue;
 pub use references::{

crates/models/src/materializations.rs

@@ -41,6 +41,13 @@ pub struct MaterializationDef {
  pub delete: bool,
 }
 
+#[derive(Serialize, Deserialize, Clone, Debug, JsonSchema, PartialEq)]
+#[serde(untagged)]
+pub enum DekafConfigContainer {
+ Direct(DekafConfig),
+ Indirect(String),
+}
+
 /// An Endpoint connector used for Flow materializations.
 #[derive(Serialize, Deserialize, Clone, Debug, JsonSchema, PartialEq)]
 #[serde(deny_unknown_fields, rename_all = "camelCase")]
@@ -51,7 +58,7 @@ pub enum MaterializationEndpoint {
  /// # A local command (development only).
  Local(LocalConfig),
  /// # A Dekaf connection
- Dekaf(DekafConfig),
+ Dekaf(DekafConfigContainer),
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, JsonSchema, PartialEq)]

crates/sources/src/indirect.rs

@@ -1,5 +1,6 @@
 use super::Format;
 use crate::Scope;
+use models::RawValue;
 use proto_flow::flow::ContentType;
 use std::collections::BTreeMap;
 
@@ -498,10 +499,28 @@ fn indirect_materialization(
  resources,
  threshold,
  ),
- // Dekaf isn't a pluggable connector, and so does not have dynamic config.
- // All of its config is defined directly within models::DekafConfig, and so
- // should not be indirected.
- models::MaterializationEndpoint::Dekaf(_) => {}
+ // I don't think this case can ever get hit as `indirect_materialization` is only called by
+ // `do_discover`, `do_pull_specs`, and `do_develop`, all of which are working with fully
+ // inlined specs.
+ models::MaterializationEndpoint::Dekaf(models::DekafConfigContainer::Indirect(_)) => {
+ tracing::warn!("Unexpectedly tried to indirect an already indirected location (dekaf)");
+ }
+ models::MaterializationEndpoint::Dekaf(models::DekafConfigContainer::Direct(config)) => {
+ indirect_dom(
+ Scope::new(scope)
+ .push_prop("endpoint")
+ .push_prop("local")
+ .push_prop("config"),
+ &mut RawValue::from_value(
+ &serde_json::to_value(config).expect("Serializing DekafConfig should not fail"),
+ ),
+ ContentType::Config,
+ format!("{base}.config"),
+ imports,
+ resources,
+ threshold,
+ )
+ }
  }
 
  for (index, models::MaterializationBinding { resource, .. }) in bindings.iter_mut().enumerate()

crates/sources/src/loader.rs

@@ -480,7 +480,7 @@ impl<F: Fetcher> Loader<F> {
  models::DeriveUsing::Connector(models::ConnectorConfig { config, .. }) => {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("using")
  .push_prop("connector")
@@ -496,7 +496,7 @@ impl<F: Fetcher> Loader<F> {
  for (index, migration) in migrations.iter().enumerate() {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("using")
  .push_prop("sqlite")
@@ -513,7 +513,7 @@ impl<F: Fetcher> Loader<F> {
  models::DeriveUsing::Typescript(models::DeriveUsingTypescript { module, .. }) => {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("using")
  .push_prop("typescript")
@@ -528,7 +528,7 @@ impl<F: Fetcher> Loader<F> {
  models::DeriveUsing::Local(models::LocalConfig { config, .. }) => {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("using")
  .push_prop("local")
@@ -545,7 +545,7 @@ impl<F: Fetcher> Loader<F> {
  for (index, transform) in spec.transforms.iter().enumerate() {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("transforms")
  .push_item(index)
@@ -560,7 +560,7 @@ impl<F: Fetcher> Loader<F> {
  if let models::Shuffle::Lambda(lambda) = &transform.shuffle {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("transforms")
  .push_item(index)
@@ -590,7 +590,7 @@ impl<F: Fetcher> Loader<F> {
  models::CaptureEndpoint::Connector(models::ConnectorConfig { config, .. }) => {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("endpoint")
  .push_prop("connector")
@@ -605,7 +605,7 @@ impl<F: Fetcher> Loader<F> {
  models::CaptureEndpoint::Local(models::LocalConfig { config, .. }) => {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("endpoint")
  .push_prop("local")
@@ -622,7 +622,7 @@ impl<F: Fetcher> Loader<F> {
  for (index, binding) in spec.bindings.iter().enumerate() {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("bindings")
  .push_item(index)
@@ -660,7 +660,7 @@ impl<F: Fetcher> Loader<F> {
  }) => {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("endpoint")
  .push_prop("connector")
@@ -675,7 +675,7 @@ impl<F: Fetcher> Loader<F> {
  models::MaterializationEndpoint::Local(models::LocalConfig { config, .. }) => {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("endpoint")
  .push_prop("local")
@@ -687,16 +687,31 @@ impl<F: Fetcher> Loader<F> {
  .boxed(),
  );
  }
- // Dekaf isn't a pluggable connector, and so does not have dynamic config to possibly
- // load from a reference. All of its config is defined directly within models::DekafConfig.
- models::MaterializationEndpoint::Dekaf(_) => {}
+ models::MaterializationEndpoint::Dekaf(models::DekafConfigContainer::Indirect(
+ location,
+ )) => {
+ tasks.push(
+ async move {
+ self.load_config(
+ scope
+ .push_prop("endpoint")
+ .push_prop("dekaf")
+ .push_prop("config"),
+ location.as_str(),
+ )
+ .await
+ }
+ .boxed(),
+ );
+ }
+ models::MaterializationEndpoint::Dekaf(models::DekafConfigContainer::Direct(_)) => {}
  };
 
  for (index, binding) in spec.bindings.iter().enumerate() {
  if !binding.disable {
  tasks.push(
  async move {
- self.load_config(
+ self.maybe_load_config(
  scope
  .push_prop("bindings")
  .push_item(index)
@@ -752,22 +767,30 @@ impl<F: Fetcher> Loader<F> {
  );
  }
 
- async fn load_config<'s>(&'s self, scope: Scope<'s>, config: &RawValue) {
-  // If `config` is a JSON string that has no whitespace then presume and
-  // require that it's a relative or absolute URL to an imported file.
+ /// If `config` is a JSON string that has no whitespace then presume and
+ /// require that it's a relative or absolute URL to an imported file.
+ async fn maybe_load_config<'s>(&'s self, scope: Scope<'s>, config: &RawValue) {
  match serde_json::from_str::<&str>(config.get()) {
  Ok(import) if !import.chars().any(char::is_whitespace) => {
- self.load_import(
- scope,
- self.fallible(scope, scope.resource().join(&import)),
- flow::ContentType::Config,
- )
- .await;
+ self.load_config(scope, import).await
  }
  _ => {}
  }
  }
 
+ /// If `config` has no whitespace then presume and require that
+ /// it's a relative or absolute URL to an imported file.
+ async fn load_config<'s>(&'s self, scope: Scope<'s>, config: &str) {
+ if !config.chars().any(char::is_whitespace) {
+ self.load_import(
+ scope,
+ self.fallible(scope, scope.resource().join(config)),
+ flow::ContentType::Config,
+ )
+ .await;
+ }
+ }
+
  // Consume a result capable of producing a LoadError.
  // Pass through a Result::Ok<T> as Some<T>.
  // Or, record a Result::Err<T> and return None.

crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap

@@ -465,6 +465,16 @@ expression: "&schema"
  "description": "Dekaf configuration. Currently empty, but present to enable easy addition of config options when they show up in the future.",
  "type": "object"
  },
+ "DekafConfigContainer": {
+ "anyOf": [
+ {
+ "$ref": "#/definitions/DekafConfig"
+ },
+ {
+ "type": "string"
+ }
+ ]
+ },
  "Derivation": {
  "description": "Derive specifies how a collection is derived from other collections.",
  "type": "object",
@@ -931,7 +941,7 @@ expression: "&schema"
  ],
  "properties": {
  "dekaf": {
- "$ref": "#/definitions/DekafConfig"
+ "$ref": "#/definitions/DekafConfigContainer"
  }
  },
  "additionalProperties": false