enhancement(pulsar sink): Refactor to use StreamSink

Cargo.toml

@@ -285,7 +285,7 @@ ordered-float = { version = "3.4.0", default-features = false }
 percent-encoding = { version = "2.2.0", default-features = false }
 pin-project = { version = "1.0.12", default-features = false }
 postgres-openssl = { version = "0.5.0", default-features = false, features = ["runtime"], optional = true }
-pulsar = { version = "5.0.0", default-features = false, features = ["tokio-runtime", "auth-oauth2"], optional = true }
+pulsar = { version = "5.0.0", default-features = false, features = ["tokio-runtime", "auth-oauth2", "lz4", "flate2", "zstd", "snap"], optional = true }
 rand = { version = "0.8.5", default-features = false, features = ["small_rng"] }
 rand_distr = { version = "0.4.3", default-features = false }
 rdkafka = { version = "0.29.0", default-features = false, features = ["tokio", "libz", "ssl", "zstd"], optional = true }
@@ -680,7 +680,7 @@ sinks-new_relic_logs = ["sinks-http"]
 sinks-new_relic = []
 sinks-papertrail = ["dep:syslog"]
 sinks-prometheus = ["aws-core", "dep:base64", "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 = "extracting_property",
+            error_type = error_type::PARSER_FAILED,
+            stage = error_stage::RECEIVING,
+            property_field = self.property_field,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            "component_errors_total", 1,
+            "error_code" => "extracing_property",
+            "error_type" => error_type::PARSER_FAILED,
+            "stage" => error_stage::PROCESSING,
+        );
+    }
+}

src/sinks/mod.rs

@@ -298,7 +298,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")]