enhancement(pulsar sink): Refactor to use StreamSink

Cargo.toml

@@ -688,7 +688,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"]
-sinks-pulsar = ["dep:apache-avro", "dep:pulsar"]
+sinks-pulsar = ["dep:apache-avro", "dep:pulsar", "dep:lru"]
 sinks-redis = ["dep:redis"]
 sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"]
 sinks-socket = ["sinks-utils-udp"]

scripts/integration/pulsar/test.yaml

@@ -1,7 +1,7 @@
 features:
 - pulsar-integration-tests
 
-test_filter: '::pulsar::'
+test_filter: '::pulsar::integration_tests::'
 
 env:
   PULSAR_ADDRESS: pulsar://pulsar:6650

src/internal_events/pulsar.rs

@@ -33,3 +33,26 @@ impl InternalEvent for PulsarSendingError {
         });
     }
 }
+
+pub struct PulsarPropertyExtractionError<F: std::fmt::Display> {
+    pub property_field: F,
+}
+
+impl<F: std::fmt::Display> InternalEvent for PulsarPropertyExtractionError<F> {
+    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::PROCESSING,
+            property_field = %self.property_field,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            "component_errors_total", 1,
+            "error_code" => "extracting_property",
+            "error_type" => error_type::PARSER_FAILED,
+            "stage" => error_stage::PROCESSING,
+        );
+    }
+}

src/sinks/mod.rs

@@ -357,7 +357,7 @@ pub enum Sinks {
     /// Publish observability events to Apache Pulsar topics.
     #[cfg(feature = "sinks-pulsar")]
     #[configurable(metadata(docs::label = "Pulsar"))]
-    Pulsar(pulsar::PulsarSinkConfig),
+    Pulsar(pulsar::config::PulsarSinkConfig),
 
     /// Publish observability data to Redis.
     #[cfg(feature = "sinks-redis")]