enhancement(pulsar sink): Refactor to use StreamSink

Cargo.toml

@@ -664,7 +664,7 @@ sinks-new_relic_logs = ["sinks-http"]
 sinks-new_relic = []
 sinks-papertrail = ["dep:syslog"]
 sinks-prometheus = ["dep:prometheus-parser", "dep:snap", "dep:serde_with"]
-sinks-pulsar = ["dep:avro-rs", "dep:pulsar"]
+sinks-pulsar = ["dep:avro-rs", "dep:pulsar", "dep:lru"]
 sinks-redis = ["dep:redis"]
 sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"]
 sinks-socket = ["sinks-utils-udp"]

src/internal_events/pulsar.rs

@@ -33,3 +33,26 @@ impl InternalEvent for PulsarSendingError {
         });
     }
 }
+
+pub struct PulsarPropertyExtractionError<'a> {
+    pub property_field: &'a str,
+}
+
+impl InternalEvent for PulsarPropertyExtractionError<'_> {
+    fn emit(self) {
+        error!(
+            message = "Failed to extract properties. Value should be a map of String -> Bytes.",
+            error_code = "extracing_property",
+            error_type = error_type::PARSER_FAILED,
+            stage = error_stage::RECEIVING,
+            property_field = self.property_field,
+            internal_log_rate_secs = 10,
+        );
+        counter!(
+            "component_errors_total", 1,
+            "error_code" => "extracing_property",
+            "error_type" => error_type::PARSER_FAILED,
+            "stage" => error_stage::RECEIVING,
+        );
+    }
+}

src/sinks/mod.rs

@@ -287,7 +287,7 @@ pub enum Sinks {
 
     /// Apache Pulsar.
     #[cfg(feature = "sinks-pulsar")]
-    Pulsar(#[configurable(derived)] pulsar::PulsarSinkConfig),
+    Pulsar(#[configurable(derived)] pulsar::config::PulsarSinkConfig),
 
     /// Redis.
     #[cfg(feature = "sinks-redis")]