Fixed the blackhole sink to properly implement end-to-end acknowledgements. Previously, the sink consumed events without updating finalizer status, causing sources that depend on acknowledgements (like`aws_s3` with SQS) to never delete processed messages from the queue.

authors: sanjams2

blackhole::{config::BlackholeConfig, sink::BlackholeSink},

test_util::{

components::run_and_assert_nonsending_sink_compliance, random_events_with_stream,

components::{SINK_TAGS, run_and_assert_sink_compliance},

random_events_with_stream,

let sink =VectorSink::Stream(Box::new(sink));

let(_input_lines, events) =random_events_with_stream(100,10,None);

run_and_assert_nonsending_sink_compliance(sink, events,&[]).await;

run_and_assert_sink_compliance(sink, events,&SINK_TAGS).await;

event::{EventArray,EventContainer},

event::{EventArray,EventContainer,EventStatus,Finalizable},

sinks::{blackhole::config::BlackholeConfig, util::StreamSink},

whileletSome(events) = input.next().await{

whileletSome(mutevents) = input.next().await{

ifletSome(rate) =self.config.rate{

let factor:f32 =1.0 / rateasf32;

let secs:f32 = factor*(events.len()asf32);

.fetch_add(message_len.get(),Ordering::AcqRel);

let finalizers = events.take_finalizers();

finalizers.update_status(EventStatus::Delivered);

events_sent.emit(CountByteSize(events.len(), message_len));

bytes_sent.emit(ByteSize(message_len.get()));