krabka_sdk_conformance/
harness.rs

1//! Subprocess adapter harness for SDK conformance vectors.
2
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    net::SocketAddr,
6    path::PathBuf,
7    process::Stdio,
8    sync::Arc,
9    time::Duration,
10};
11
12use krabka_broker::{Broker, BrokerConfig, BrokerHandle};
13use krabka_client_admin::{AdminClient, CreateTopicSpec};
14use krabka_client_core::Client;
15use krabka_gateway::{
16    codec::{CodecError, Decoded, EncodeBody, RecordCodec},
17    config::GatewayConfig,
18    produce::ProduceCore,
19    serve,
20    state::AppState,
21};
22use krabka_protocol::owned::incremental_alter_configs_request::{
23    AlterConfigsResource, AlterableConfig, IncrementalAlterConfigsRequest,
24};
25use krabka_units::prelude::*;
26use tokio::{
27    io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader, BufWriter},
28    net::TcpListener,
29    process::{Child, ChildStdin, ChildStdout, Command as ProcessCommand},
30    task::JoinHandle,
31};
32use tokio_util::sync::CancellationToken;
33
34use crate::{
35    protocol::{CONTRACT_MAJOR, Command, Response},
36    vectors::{ContractVersion, Vector, VectorError, load_vectors},
37};
38
39const LIVE_COMPATIBLE_VECTOR_IDS: &[&str] = &[
40    "messaging_roundtrip",
41    "ce_binary_mapping",
42    "filter_delivers_matches_only",
43    "header_shape",
44    "queue_v1_1_ack_error_shape",
45    "queue_v1_1_ack_shape",
46    "queue_v1_1_acquire_error_shape",
47    "queue_v1_1_acquire_shape",
48    "queue_v1_1_live_error_mapping",
49    "queue_v1_1_lock_duration_error_shape",
50    "queue_v1_1_renew_shape",
51    "queue_v1_1_session_ownership",
52];
53const ADAPTER_CALL_TIMEOUT: Duration = Duration::from_secs(30);
54
55/// Harness configuration.
56#[derive(Debug, Clone)]
57pub struct HarnessConfig {
58    /// Adapter executable path.
59    pub adapter: PathBuf,
60    /// Adapter executable arguments.
61    pub adapter_args: Vec<String>,
62    /// Vector directory.
63    pub vectors_dir: PathBuf,
64    /// Optional vector id filter.
65    pub filter: Option<String>,
66    /// Gateway endpoint sent through `Configure`.
67    pub endpoint: String,
68    /// Substrate used by the harness.
69    pub substrate: HarnessSubstrate,
70    /// Run only explicit live-only vectors supported by the current live Rust app SDK.
71    pub live_compatible_only: bool,
72}
73
74/// Substrate used for adapter calls.
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
76pub enum HarnessSubstrate {
77    /// Use the configured endpoint as-is.
78    External,
79    /// Boot an in-process broker and plaintext h2c gateway on `127.0.0.1:0`.
80    Live,
81}
82
83/// SDK conformance harness.
84#[derive(Debug, Clone)]
85pub struct Harness {
86    config: HarnessConfig,
87}
88
89impl Harness {
90    /// Create a harness.
91    #[must_use]
92    pub fn new(config: HarnessConfig) -> Self {
93        Self { config }
94    }
95
96    /// Load vectors and run them against the adapter.
97    ///
98    /// # Errors
99    ///
100    /// Returns an error when vectors cannot be loaded, the adapter or live
101    /// substrate cannot start, adapter I/O fails, or shutdown times out.
102    pub async fn run(&self) -> Result<RunSummary, HarnessError> {
103        let adapter_version = self.discover_adapter_version().await?;
104        let mut vectors = load_vectors(&self.config.vectors_dir)?;
105        let mut skipped = newer_contract_skips(adapter_version, &vectors);
106        vectors.retain(|vector| adapter_version.satisfies(vector));
107        if let Some(filter) = &self.config.filter {
108            vectors.retain(|vector| vector.id == *filter);
109            skipped.retain(|skipped| skipped.vector_id == *filter);
110        }
111        if self.config.substrate != HarnessSubstrate::Live {
112            let live_only = vectors
113                .iter()
114                .filter(|vector| vector.live_only)
115                .map(|vector| SkippedVector {
116                    vector_id: vector.id.clone(),
117                    reason: "requires live substrate".into(),
118                })
119                .collect::<Vec<_>>();
120            vectors.retain(|vector| !vector.live_only);
121            skipped.extend(live_only);
122        }
123        let live_plan = if self.config.live_compatible_only {
124            LiveCompatiblePlan::from_full_vectors(vectors)
125        } else {
126            LiveCompatiblePlan::full_vectors(vectors)
127        };
128        skipped.extend(live_plan.skipped);
129        match self.config.substrate {
130            HarnessSubstrate::External => {
131                self.run_vectors_with_endpoint(live_plan.vectors, skipped, &self.config.endpoint)
132                    .await
133            }
134            HarnessSubstrate::Live => {
135                let topic_names = topic_names_for_vectors(&live_plan.vectors);
136                let queue_groups = queue_groups_for_vectors(&live_plan.vectors);
137                let live = LiveSubstrate::boot(&topic_names, &queue_groups).await?;
138                let endpoint = live.endpoint.clone();
139                let result = self
140                    .run_vectors_with_endpoint(live_plan.vectors, skipped, &endpoint)
141                    .await;
142                live.shutdown().await?;
143                result
144            }
145        }
146    }
147
148    /// Run already-loaded vectors.
149    ///
150    /// # Errors
151    ///
152    /// Returns an error when the adapter cannot start, adapter I/O fails, or an
153    /// adapter call times out.
154    pub async fn run_vectors(&self, vectors: Vec<Vector>) -> Result<RunSummary, HarnessError> {
155        self.run_vectors_with_endpoint(vectors, vec![], &self.config.endpoint)
156            .await
157    }
158
159    async fn run_vectors_with_endpoint(
160        &self,
161        vectors: Vec<Vector>,
162        skipped: Vec<SkippedVector>,
163        endpoint: &str,
164    ) -> Result<RunSummary, HarnessError> {
165        let mut passed = 0;
166        let mut failed = vec![];
167        for vector in vectors {
168            match self.run_vector(&vector, endpoint).await? {
169                Some(failure) => failed.push(failure),
170                None => passed += 1,
171            }
172        }
173        Ok(RunSummary {
174            passed,
175            failed,
176            skipped,
177        })
178    }
179
180    async fn run_vector(
181        &self,
182        vector: &Vector,
183        endpoint: &str,
184    ) -> Result<Option<VectorFailure>, HarnessError> {
185        let mut adapter = AdapterProcess::spawn(&self.config.adapter, &self.config.adapter_args)?;
186        let hello = adapter.call(Command::Hello).await?;
187        match hello {
188            Response::Hello {
189                contract_major,
190                contract_minor,
191                ..
192            } if ContractVersion::new(contract_major, contract_minor).satisfies(vector) => {}
193            actual => {
194                let expected = Response::Hello {
195                    contract_major: CONTRACT_MAJOR,
196                    contract_minor: vector.since.minor,
197                    language: "<adapter>".into(),
198                };
199                return Ok(Some(VectorFailure {
200                    vector_id: vector.id.clone(),
201                    step: "hello".into(),
202                    expected,
203                    actual,
204                }));
205            }
206        }
207        let configured = adapter
208            .call(Command::Configure {
209                endpoint: endpoint.to_string(),
210                bearer: None,
211            })
212            .await?;
213        let expected_configuration =
214            Response::Ok(serde_json::json!({ "bearer_configured": false }));
215        if configured != expected_configuration {
216            return Ok(Some(VectorFailure {
217                vector_id: vector.id.clone(),
218                step: "configure".into(),
219                expected: expected_configuration,
220                actual: configured,
221            }));
222        }
223        for step in &vector.steps {
224            let actual = adapter.call(step.command.clone()).await?;
225            if actual != step.expect {
226                return Ok(Some(VectorFailure {
227                    vector_id: vector.id.clone(),
228                    step: step.name.clone(),
229                    expected: step.expect.clone(),
230                    actual,
231                }));
232            }
233        }
234        Ok(None)
235    }
236
237    async fn discover_adapter_version(&self) -> Result<ContractVersion, HarnessError> {
238        let mut adapter = AdapterProcess::spawn(&self.config.adapter, &self.config.adapter_args)?;
239        let hello = adapter.call(Command::Hello).await?;
240        let Response::Hello {
241            contract_major,
242            contract_minor,
243            language: _,
244        } = hello
245        else {
246            return Err(HarnessError::AdapterProtocol(
247                "adapter hello did not return a hello response",
248            ));
249        };
250        if contract_major != CONTRACT_MAJOR {
251            return Err(HarnessError::AdapterProtocol(
252                "adapter contract major is not supported",
253            ));
254        }
255        Ok(ContractVersion::new(contract_major, contract_minor))
256    }
257}
258
259fn newer_contract_skips(
260    adapter_version: ContractVersion,
261    vectors: &[Vector],
262) -> Vec<SkippedVector> {
263    vectors
264        .iter()
265        .filter(|vector| adapter_version < vector.since)
266        .map(|vector| SkippedVector {
267            vector_id: vector.id.clone(),
268            reason: format!(
269                "requires contract {}.{}; adapter declares {}.{}",
270                vector.since.major,
271                vector.since.minor,
272                adapter_version.major,
273                adapter_version.minor
274            ),
275        })
276        .collect()
277}
278
279struct LiveCompatiblePlan {
280    vectors: Vec<Vector>,
281    skipped: Vec<SkippedVector>,
282}
283
284impl LiveCompatiblePlan {
285    fn full_vectors(vectors: Vec<Vector>) -> Self {
286        Self {
287            vectors,
288            skipped: vec![],
289        }
290    }
291
292    fn from_full_vectors(vectors: Vec<Vector>) -> Self {
293        let mut live_vectors = Vec::new();
294        let mut skipped = Vec::new();
295
296        for vector in vectors {
297            if !LIVE_COMPATIBLE_VECTOR_IDS.contains(&vector.id.as_str()) {
298                skipped.push(SkippedVector {
299                    vector_id: vector.id,
300                    reason: "not supported by live substrate".into(),
301                });
302                continue;
303            }
304            live_vectors.push(vector);
305        }
306
307        Self {
308            vectors: live_vectors,
309            skipped,
310        }
311    }
312}
313
314fn topic_names_for_vectors(vectors: &[Vector]) -> Vec<String> {
315    let mut names = BTreeSet::new();
316    for step in vectors.iter().flat_map(|vector| &vector.steps) {
317        let topic = match &step.command {
318            Command::Publish { topic, .. } | Command::PublishEvent { topic, .. } => topic,
319            Command::Subscribe { topics, .. } => {
320                for topic in topics {
321                    if is_creatable_topic(topic) {
322                        names.insert(topic.clone());
323                    }
324                }
325                continue;
326            }
327            _ => continue,
328        };
329        if is_creatable_topic(topic) {
330            names.insert(topic.clone());
331        }
332    }
333    names.into_iter().collect()
334}
335
336/// The share groups that the vectors acquire from, each named once.
337///
338/// An empty group name is left out. The error vectors send one on purpose, and
339/// the broker refuses a config for it.
340fn queue_groups_for_vectors(vectors: &[Vector]) -> Vec<String> {
341    vectors
342        .iter()
343        .flat_map(|vector| &vector.steps)
344        .filter_map(|step| match &step.command {
345            Command::QueueAcquire { group, .. } if !group.is_empty() => Some(group.clone()),
346            _ => None,
347        })
348        .collect::<BTreeSet<_>>()
349        .into_iter()
350        .collect()
351}
352
353fn is_creatable_topic(topic: &str) -> bool {
354    !topic.is_empty() && !topic.starts_with("__missing_")
355}
356
357struct LiveSubstrate {
358    endpoint: String,
359    broker: BrokerHandle,
360    shutdown: CancellationToken,
361    gateway_task: JoinHandle<std::io::Result<()>>,
362    _data_dir: tempfile::TempDir,
363}
364
365impl LiveSubstrate {
366    async fn boot(topic_names: &[String], queue_groups: &[String]) -> Result<Self, HarnessError> {
367        let data_dir = tempfile::TempDir::new()?;
368        let mut broker_config = BrokerConfig::for_tests(data_dir.path().to_path_buf());
369        broker_config.classic_group_initial_rebalance_delay = millis(1);
370        // Each SDK gives an acquisition one second. A partition initialized
371        // after the join must become assigned on a heartbeat within that budget.
372        broker_config.share_group.heartbeat_interval = Duration::from_millis(100);
373        broker_config.share_group.min_heartbeat_interval = Duration::from_millis(100);
374        broker_config.share_group.max_heartbeat_interval = Duration::from_millis(200);
375        // `for_tests` names the controller as `127.0.0.1:0`. The broker binds
376        // an ephemeral port for it, but its heartbeat client dials the address
377        // in `controller_quorum_voters`, which still says port 0. No heartbeat
378        // arrives, so two seconds after start the controller marks the broker
379        // dead, fences it, and leaves its partitions without a leader. Binding
380        // the controller port first gives the voter entry the real address.
381        let controller_listener = TcpListener::bind("127.0.0.1:0").await?;
382        let controller_addr = controller_listener.local_addr()?;
383        broker_config.controller_listen_addr = controller_addr;
384        broker_config.controller_quorum_voters =
385            vec![(broker_config.node_id, controller_addr.to_string())];
386        let broker =
387            Broker::start_with_controller_listener(broker_config, Some(controller_listener))
388                .await
389                .map_err(HarnessError::BrokerStart)?;
390        broker.wait_until_group_coordinator_ready().await;
391        // The first share fetch creates `__share_group_state` and waits for
392        // this broker's share coordinator to load it. On a slow host that
393        // cold start alone outlasts the one-second wait that each SDK gives an
394        // acquisition, so the first queue vector would acquire nothing. Warm
395        // the share coordinator here, as the group coordinator is warmed above.
396        if !queue_groups.is_empty() {
397            broker.wait_until_share_coordinator_ready().await;
398        }
399        let bootstrap = broker.listen_addr().to_string();
400        create_topics(&bootstrap, topic_names).await?;
401        set_share_groups_earliest(&bootstrap, queue_groups).await?;
402
403        let listener = TcpListener::bind("127.0.0.1:0").await?;
404        let listen_addr = listener.local_addr()?;
405        let state = gateway_state(&bootstrap, listen_addr).await?;
406        let app = krabka_gateway::router(state);
407        let shutdown = CancellationToken::new();
408        let gateway_shutdown = shutdown.clone();
409        let gateway_task =
410            tokio::spawn(async move { serve::serve(listener, app, None, gateway_shutdown).await });
411
412        Ok(Self {
413            endpoint: format!("http://{listen_addr}"),
414            broker,
415            shutdown,
416            gateway_task,
417            _data_dir: data_dir,
418        })
419    }
420
421    async fn shutdown(self) -> Result<(), HarnessError> {
422        self.shutdown.cancel();
423        tokio::time::timeout(Duration::from_secs(5), self.gateway_task)
424            .await
425            .map_err(|_| HarnessError::SubstrateTimeout)??
426            .map_err(HarnessError::GatewayServe)?;
427        self.broker.shutdown().await;
428        Ok(())
429    }
430}
431
432async fn create_topics(bootstrap: &str, topic_names: &[String]) -> Result<(), HarnessError> {
433    if topic_names.is_empty() {
434        return Ok(());
435    }
436    let mut admin = AdminClient::connect(&[bootstrap.to_string()])
437        .await
438        .map_err(HarnessError::Admin)?;
439    let specs = topic_names
440        .iter()
441        .map(|name| CreateTopicSpec {
442            replica_assignments: std::collections::BTreeMap::new(),
443            name: name.clone(),
444            partitions: 1,
445            replicas: 1,
446            configs: BTreeMap::new(),
447        })
448        .collect::<Vec<_>>();
449    admin
450        .create_topics(
451            &specs,
452            krabka_client_admin::TopicMutationOptions::with_timeout(millis(10_000)),
453        )
454        .await
455        .map(|_| ())
456        .map_err(HarnessError::Admin)
457}
458
459/// Kafka resource type id for `GROUP`.
460const RESOURCE_TYPE_GROUP: i8 = 32;
461
462/// `SET` in the `IncrementalAlterConfigs` wire protocol.
463const CONFIG_OPERATION_SET: i8 = 0;
464
465/// Puts each share group on `share.auto.offset.reset=earliest`.
466///
467/// The queue vectors publish a record and then acquire it. KIP-932 starts a
468/// share partition that has no state at `latest` by default, so without this
469/// setting the first acquire does not see a record published before it. An
470/// operator runs the same `kafka-configs --entity-type groups --alter` to make
471/// a share group read a topic from its beginning.
472async fn set_share_groups_earliest(bootstrap: &str, groups: &[String]) -> Result<(), HarnessError> {
473    if groups.is_empty() {
474        return Ok(());
475    }
476    let client = Client::builder()
477        .bootstrap(bootstrap.to_string())
478        .build()
479        .await
480        .map_err(HarnessError::Client)?;
481    let response = client
482        .send(IncrementalAlterConfigsRequest {
483            resources: groups
484                .iter()
485                .map(|group| AlterConfigsResource {
486                    resource_type: RESOURCE_TYPE_GROUP,
487                    resource_name: group.clone(),
488                    configs: vec![AlterableConfig {
489                        name: "share.auto.offset.reset".into(),
490                        config_operation: CONFIG_OPERATION_SET,
491                        value: Some("earliest".into()),
492                        ..Default::default()
493                    }],
494                    ..Default::default()
495                })
496                .collect(),
497            ..Default::default()
498        })
499        .await
500        .map_err(HarnessError::Client)?;
501    if let Some(rejected) = response
502        .responses
503        .into_iter()
504        .find(|resource| resource.error_code != 0)
505    {
506        return Err(HarnessError::GroupConfig {
507            group: rejected.resource_name,
508            code: rejected.error_code,
509            message: rejected.error_message.unwrap_or_default(),
510        });
511    }
512    Ok(())
513}
514
515async fn gateway_state(
516    bootstrap: &str,
517    listen_addr: SocketAddr,
518) -> Result<Arc<AppState>, HarnessError> {
519    let codec = Arc::new(ConformanceCodec);
520    let produce = ProduceCore::new(bootstrap, "sdk-conformance", codec.clone(), None)
521        .await
522        .map_err(HarnessError::GatewayInit)?;
523    let config = Arc::new(gateway_config(bootstrap, listen_addr));
524    Ok(Arc::new(AppState {
525        produce: Arc::new(produce),
526        config,
527        authz: Arc::new(krabka_gateway::authz::GatewayAuthz::new(Arc::new(
528            krabka_authz::AllowAllAuthorizer,
529        ))),
530        codec,
531        queue: Arc::new(krabka_gateway::queue::QueueSessionTable::default()),
532    }))
533}
534
535#[derive(Debug)]
536struct ConformanceCodec;
537
538#[async_trait::async_trait]
539impl RecordCodec for ConformanceCodec {
540    async fn encode(&self, _topic: &str, body: EncodeBody) -> Result<bytes::Bytes, CodecError> {
541        Ok(match body {
542            EncodeBody::Raw(value) => value,
543            EncodeBody::Structured { json, .. } => json,
544        })
545    }
546
547    async fn decode(&self, _topic: &str, value: bytes::Bytes) -> Result<Decoded, CodecError> {
548        let json = serde_json::from_slice::<serde_json::Value>(&value)
549            .ok()
550            .map(|_| value.clone());
551        Ok(Decoded {
552            value,
553            schema: None,
554            json,
555        })
556    }
557}
558
559fn gateway_config(bootstrap: &str, listen_addr: SocketAddr) -> GatewayConfig {
560    GatewayConfig {
561        bootstrap: bootstrap.to_string(),
562        listen_addr,
563        client_id: "sdk-conformance".into(),
564        dedup_topic: "__krabka_grpc_dedup".into(),
565        dedup_partitions: 4,
566        dedup_window: hours(1),
567        dedup_ownership_group: "sdk-conformance-owners".into(),
568        dedup_txn_id_prefix: "sdk-conformance-dedup".into(),
569        advertised_addr: listen_addr.to_string(),
570        membership_topic: "__krabka_gateway_membership".into(),
571        tls: None,
572        broker_security: None,
573        authz: None,
574        webhooks: BTreeMap::new().into_iter().collect(),
575        outbound: Vec::new(),
576        schema_registry_url: None,
577        runtime: krabka_gateway::config::GatewayRuntimeConfig::default(),
578    }
579}
580
581/// Summary returned by a conformance run.
582#[derive(Debug, Clone, PartialEq)]
583pub struct RunSummary {
584    /// Number of vectors that passed.
585    pub passed: usize,
586    /// Failed vector diagnostics.
587    pub failed: Vec<VectorFailure>,
588    /// Vectors intentionally excluded from this run.
589    pub skipped: Vec<SkippedVector>,
590}
591
592impl RunSummary {
593    /// Whether every vector passed.
594    #[must_use]
595    pub fn is_success(&self) -> bool {
596        self.failed.is_empty()
597    }
598}
599
600/// One vector excluded from the run.
601#[derive(Debug, Clone, PartialEq, Eq)]
602pub struct SkippedVector {
603    /// Vector id.
604    pub vector_id: String,
605    /// Human-readable reason.
606    pub reason: String,
607}
608
609/// One vector mismatch.
610#[derive(Debug, Clone, PartialEq)]
611pub struct VectorFailure {
612    /// Vector id.
613    pub vector_id: String,
614    /// Step name.
615    pub step: String,
616    /// Expected response.
617    pub expected: Response,
618    /// Actual response.
619    pub actual: Response,
620}
621
622struct AdapterProcess {
623    child: Child,
624    stdin: BufWriter<ChildStdin>,
625    stdout: BufReader<ChildStdout>,
626}
627
628impl AdapterProcess {
629    fn spawn(adapter: &PathBuf, args: &[String]) -> Result<Self, HarnessError> {
630        let mut child = ProcessCommand::new(adapter)
631            .args(args)
632            .stdin(Stdio::piped())
633            .stdout(Stdio::piped())
634            .spawn()
635            .map_err(HarnessError::Spawn)?;
636        let stdin = child
637            .stdin
638            .take()
639            .ok_or(HarnessError::MissingPipe("stdin"))?;
640        let stdout = child
641            .stdout
642            .take()
643            .ok_or(HarnessError::MissingPipe("stdout"))?;
644        Ok(Self {
645            child,
646            stdin: BufWriter::new(stdin),
647            stdout: BufReader::new(stdout),
648        })
649    }
650
651    async fn call(&mut self, command: Command) -> Result<Response, HarnessError> {
652        let line = serde_json::to_string(&command)?;
653        self.stdin.write_all(line.as_bytes()).await?;
654        self.stdin.write_all(b"\n").await?;
655        self.stdin.flush().await?;
656        let mut response = String::new();
657        tokio::time::timeout(ADAPTER_CALL_TIMEOUT, self.stdout.read_line(&mut response))
658            .await
659            .map_err(|_| HarnessError::AdapterTimeout)??;
660        if response.is_empty() {
661            return Err(HarnessError::AdapterEof);
662        }
663        Ok(serde_json::from_str(response.trim_end())?)
664    }
665}
666
667impl Drop for AdapterProcess {
668    fn drop(&mut self) {
669        let _ = self.child.start_kill();
670    }
671}
672
673/// Harness errors.
674#[derive(Debug, thiserror::Error)]
675#[non_exhaustive]
676pub enum HarnessError {
677    /// Vector load failed.
678    #[error("vectors: {0}")]
679    Vectors(#[from] VectorError),
680    /// Adapter process failed to spawn.
681    #[error("spawn adapter: {0}")]
682    Spawn(std::io::Error),
683    /// Adapter pipe was not available.
684    #[error("adapter missing {0} pipe")]
685    MissingPipe(&'static str),
686    /// Adapter I/O failed.
687    #[error("adapter io: {0}")]
688    Io(#[from] std::io::Error),
689    /// Protocol JSON failed.
690    #[error("adapter json: {0}")]
691    Json(#[from] serde_json::Error),
692    /// Adapter did not answer in time.
693    #[error("adapter timed out")]
694    AdapterTimeout,
695    /// Adapter exited without a response.
696    #[error("adapter exited before responding")]
697    AdapterEof,
698    /// Adapter violated the conformance protocol.
699    #[error("adapter protocol: {0}")]
700    AdapterProtocol(&'static str),
701    /// Broker failed to start.
702    #[error("live substrate broker start: {0}")]
703    BrokerStart(krabka_broker::BrokerError),
704    /// Admin client setup failed.
705    #[error("live substrate admin: {0}")]
706    Admin(krabka_client_admin::AdminError),
707    /// A raw protocol client request failed.
708    #[error("live substrate client: {0}")]
709    Client(krabka_client_core::ClientError),
710    /// The broker rejected a share group config.
711    #[error("live substrate group config for {group}: error {code}: {message}")]
712    GroupConfig {
713        /// Share group name.
714        group: String,
715        /// Kafka error code.
716        code: i16,
717        /// Broker error message.
718        message: String,
719    },
720    /// Gateway state failed to initialize.
721    #[error("live substrate gateway init: {0}")]
722    GatewayInit(krabka_gateway::error::GatewayError),
723    /// Gateway server returned an error.
724    #[error("live substrate gateway serve: {0}")]
725    GatewayServe(std::io::Error),
726    /// Gateway task join failed.
727    #[error("live substrate gateway task: {0}")]
728    GatewayTask(#[from] tokio::task::JoinError),
729    /// Live substrate did not shut down in time.
730    #[error("live substrate timed out")]
731    SubstrateTimeout,
732}
krabka_sdk_conformance/harness.rs

krabka_sdk_conformance/
harness.rs
