update kafka-sink service to set Errored status on events

currently all kafka errors result it a "Rejected" event, which for the
HTTP sink returns a 400 to the user.

however most errors should be "Errored", which implies an error happened
and not a bad event that can't be sent to the sink

src/sinks/kafka/service.rs

@@ -14,6 +14,7 @@ use rdkafka::{
     producer::{FutureProducer, FutureRecord},
     types::RDKafkaErrorCode,
 };
+use vector_lib::config;
 
 use crate::{kafka::KafkaStatisticsContext, sinks::prelude::*};
 
@@ -34,11 +35,12 @@ pub struct KafkaRequestMetadata {
 pub struct KafkaResponse {
     event_byte_size: GroupedCountByteSize,
     raw_byte_size: usize,
+    event_status: EventStatus,
 }
 
 impl DriverResponse for KafkaResponse {
     fn event_status(&self) -> EventStatus {
-        EventStatus::Delivered
+        self.event_status
     }
 
     fn events_sent(&self) -> &GroupedCountByteSize {
@@ -153,6 +155,7 @@ impl Service<KafkaRequest> for KafkaService {
                             .map(|_| KafkaResponse {
                                 event_byte_size,
                                 raw_byte_size,
+                                event_status: EventStatus::Delivered,
                             })
                             .map_err(|(err, _)| err);
                     }
@@ -168,8 +171,30 @@ impl Service<KafkaRequest> for KafkaService {
                         record = original_record;
                         tokio::time::sleep(Duration::from_millis(100)).await;
                     }
-                    // A different error occurred.
-                    Err((err, _)) => return Err(err),
+                    // A final/unretriable error occurred.
+                    Err((
+                        err @ KafkaError::MessageProduction(
+                            RDKafkaErrorCode::InvalidMessage |
+                            RDKafkaErrorCode::InvalidMessageSize |
+                            RDKafkaErrorCode::MessageSizeTooLarge |
+                            RDKafkaErrorCode::UnknownTopicOrPartition |
+                            RDKafkaErrorCode::InvalidRecord |
+                            RDKafkaErrorCode::InvalidRequiredAcks |
+                            RDKafkaErrorCode::TopicAuthorizationFailed |
+                            RDKafkaErrorCode::UnsupportedForMessageFormat |
+                            RDKafkaErrorCode::ClusterAuthorizationFailed
+                        ),
+                        _,
+                    )) => return Err(err),
+
+                    // A different error occurred. Set event status to Errored not Rejected.
+                    Err(_) => {
+                        return Ok(KafkaResponse {
+                            event_byte_size: config::telemetry().create_request_count_byte_size(),
+                            raw_byte_size: 0,
+                            event_status: EventStatus::Errored,
+                        })
+                    }
                 };
             }
         })