Merge remote-tracking branch 'origin/main' into refactor

Author: Devdutt Shenoi

src/connectors/kafka/processor.rs

@@ -81,8 +81,7 @@ impl ParseableSinkProcessor {
             custom_partition_values: HashMap::new(),
             stream_type: StreamType::UserDefined,
         }
-        .process(&stream)
-        .await?;
+        .process(&stream)?;
 
         Ok(())
     }

src/event/mod.rs

@@ -46,7 +46,7 @@ pub struct Event {
 
 // Events holds the schema related to a each event for a single log stream
 impl Event {
-    pub async fn process(self, stream: &Stream) -> Result<(), EventError> {
+    pub fn process(self, stream: &Stream) -> Result<(), EventError> {
         let mut key = get_schema_key(&self.rb.schema().fields);
         if self.time_partition.is_some() {
             let parsed_timestamp_to_min = self.parsed_timestamp.format("%Y%m%dT%H%M").to_string();

src/handlers/http/ingest.rs

@@ -99,8 +99,8 @@ pub async fn ingest_internal_stream(stream_name: String, body: Bytes) -> Result<
         custom_partition_values: HashMap::new(),
         stream_type: StreamType::Internal,
     }
-    .process(&stream)
-    .await?;
+    .process(&stream)?;
+
     Ok(())
 }
 

src/otel/traces.rs

@@ -293,9 +293,15 @@ fn flatten_span_record(span_record: &Span) -> Vec<Map<String, Value>> {
         span_record_json.extend(flatten_status(status));
     }
 
-    for span_json in &mut span_records_json {
-        for (key, value) in &span_record_json {
-            span_json.insert(key.clone(), value.clone());
+    // if span_record.events is null, code should still flatten other elements in the span record - this is handled in the if block
+    // else block handles the flattening the span record that includes events and links records in each span record
+    if span_records_json.is_empty() {
+        span_records_json = vec![span_record_json];
+    } else {
+        for span_json in &mut span_records_json {
+            for (key, value) in &span_record_json {
+                span_json.insert(key.clone(), value.clone());
+            }
         }
     }
 

src/parseable/streams.rs

@@ -216,9 +216,9 @@ impl Stream {
                 custom_partition_values,
                 stream_type: StreamType::UserDefined,
             }
-            .process(self)
-            .await?;
+            .process(self)?;
         }
+
         Ok(())
     }