async fn test(
        op: impl Into<DmlOperation> + Send,
        metrics: &metric::Registry,
        with_sink_return: Result<bool, crate::data::Error>,
        with_fetcher_return: Option<u64>,
    ) -> Result<bool, crate::data::Error> {
        let op = op.into();
        let inner = MockDmlSink::default().with_apply_return([with_sink_return]);
        let instrumentation = SinkInstrumentation::new(
            inner,
            MockWatermarkFetcher::new(with_fetcher_return),
            TEST_KAFKA_TOPIC.to_string(),
            KafkaPartition::new(SEQUENCER_ID as _),
            metrics,
        );

        instrumentation.apply(op).await
    }
/// Initialise a [`SinkInstrumentation`] and drive it with the given
/// parameters.
Initialise a [`SinkInstrumentation`] and drive it with the given
parameters.
Initialise a [ ` SinkInstrumentation ` ] and drive it with the given parameters .
async fn test(
        op: impl Into<DmlOperation> + Send,
        metrics: &metric::Registry,
        with_sink_return: Result<bool, crate::data::Error>,
        with_fetcher_return: Option<u64>,
    ) -> Result<bool, crate::data::Error> {
        let op = op.into();
        let inner = MockDmlSink::default().with_apply_return([with_sink_return]);
        let instrumentation = SinkInstrumentation::new(
            inner,
            MockWatermarkFetcher::new(with_fetcher_return),
            TEST_KAFKA_TOPIC.to_string(),
            KafkaPartition::new(SEQUENCER_ID as _),
            metrics,
        );
        instrumentation.apply(op).await
    }
async fn test ( op : impl Into < DmlOperation > + Send , metrics : & metric :: Registry , with_sink_return : Result < bool , crate :: data :: Error > , with_fetcher_return : Option < u64 > , ) -> Result < bool , crate :: data :: Error > { let op = op . into ( ) ; let inner = MockDmlSink :: default ( ) . with_apply_return ( [ with_sink_return ] ) ; let instrumentation = SinkInstrumentation :: new ( inner , MockWatermarkFetcher :: new ( with_fetcher_return ) , TEST_KAFKA_TOPIC . to_string ( ) , KafkaPartition :: new ( SEQUENCER_ID as _ ) , metrics , ) ; instrumentation . apply ( op ) . await }
Initialise a [`SinkInstrumentation`] and drive it with the given
parameters.
Initialise a [ ` SinkInstrumentation ` ] and drive it with the given parameters .
op
impl Into<DmlOperation> + Send
metrics
&metric::Registry
with_sink_return
Result<bool, crate::data::Error>
with_fetcher_return
Option<u64>
op
impl Into<DmlOperation> + Send
metrics
&metric::Registry
with_sink_return
Result<bool, crate::data::Error>
with_fetcher_return
Option<u64>