Dekaf runtime changes #1854
Merged
Dekaf runtime changes #1854
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
We need `runtime` to depend on `dekaf`, and they both need `unseal`, so it needed to be moved somewhere, and there didn't seem to be anywhere good to put it other than its own crate.
In order for field selection to work in the UI, it needs to be able to validate a draft with a Dekaf connector in it. This detects when the request is for a Dekaf materialization and re-routes it to Dekaf's `unary_materialize`. It also means that the `connector_tags` job can work for Dekaf with a small change to detect image names starting in `ghcr.io/estuary/dekaf-*` and mark them as `ConnectorType::Dekaf`.
psFried
approved these changes
Jan 6, 2025
crates/agent/src/connector_tags.rs
Outdated
| Err(err) => { | ||
| tracing::warn!(image = %image_composed, error = %err, "failed to determine connector protocol"); | ||
| return Ok((row.tag_id, JobStatus::SpecFailed)); | ||
| let proto_type = if row.image_name.starts_with(DEKAF_IMAGE_NAME_PREFIX) { |
There was a problem hiding this comment.
I kinda feel like this conditional should be in runtime::flow_runtime_protocol. Is there a reason to prefer it to be in agent?
There was a problem hiding this comment.
Yeah, you're right. I read runtime::flow_runtime_protocol and parse_image_inspection briefly and saw that docker inspect call, but it's trivial to check the image first. 👍
4 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description:
Rolls up a couple more changes to the agent and runtime for Dekaf:
Runtime::unary_materialize()and route them todekaf::connector::unary_materialize(). Doing it here instead of inbuildallows more of Flow to interact with Dekaf tasks correctly:dry_runpublication (i.e avalidatecall) to fetch which fields to show. This now intercepts that call and correctly passes it to Dekaf because we setconnector_typebased on the endpoint config inwalk_materialization():flow/crates/validation/src/materialization.rs
Lines 126 to 128 in 44dc0db
connector_tagsagent task calls a connector'sSpecRPC, and we need to get Dekaf's endpoint/resource config schemas somehow so I figured rather than writing a thing to do it manually, all we need to do is sendconnector_type: ConnectorType::Dekafwhen we detect a task withimage_namestarting inghcr.io/estuary/dekaf-.connector_tagswhen processing a Dekaf row as obviously there's no image to pull.runtimeto depend ondekaf, and they both needunsealso it needed to be moved somewhere, and there didn't seem to be anywhere good to put it other than its own crate.This change is