runtime: Correct handling for calling dekaf::connector::unary_materialize

[jshearer]

Description:

Originally the Dekaf special-case lived in Runtime::unary_materialize, but #2001 changed the way unary RPCs are invoked in a way that bypassed that logic. Moving it into Runtime::serve_materialize fixes things.

---

This change is 

[jgraettinger]

For a better way to do this, please take a look at how derive-sqlite is implemented:

https://github.com/estuary/flow/blob/master/crates/runtime/src/derive/connector.rs#L91-L94

You should be able to implement an equivalent connector stream for Dekaf and just call it from materialize::connector::start(), without needing to be aware of it elsewhere

[jgraettinger]

You don't actually need to spawn it in a tokio task -- those are details that are specific to how sqlite needs to work.

You can instead model off of the validation noop implementation. It only needs to serve Spec and Validate, and can error on anything else.

[jshearer]

Ok, the noop implementation makes sense -- basically just a coroutine loop that passes Spec and Validate calls to dekaf::connector::unary_materialize() and errors otherwise.

[jgraettinger]

I would get rid of dekaf::connector::unary_materialize() altogether, and implement a dekaf connector stream in terms of coroutine that services Spec and Validate, and errors on anything else.

I'd expect dekaf::connector(connector_rx) to return an impl Stream of materialization responses, pretty much identically to how derive_sqlite::connector(connector_rx) works, except you can implement it using a simple coroutine rather than a tokio-spawned task.

[jshearer]

That turned out nice and clean! Best reviewed with ?w=1 due to whitespace-only changes

[jgraettinger]

LGTM