Send ack/fail if event.transactional for otel sink.
@@ -136,12 +136,12 @@ impl Sink for OpenTelemetry {
error!("Failed to dispatch otel/gRPC metrics message: {}", e);
self.is_down = true;
if event.transactional {
- return Ok(Some(vec![qos::close(&mut event)]));
- }
return Ok(Some(vec![
qos::fail(&mut event.clone()),
qos::close(&mut event),
]));
+ }
+ return Ok(Some(vec![qos::close(&mut event)]));
};
continue;
}
@@ -153,13 +153,13 @@ impl Sink for OpenTelemetry {
error!("Failed to dispatch otel/gRPC logs message: {}", e);
self.is_down = true;
if event.transactional {
- return Ok(Some(vec![qos::close(&mut event)]));
- }
return Ok(Some(vec![
qos::fail(&mut event.clone()),
qos::close(&mut event),
]));
}
+ return Ok(Some(vec![qos::close(&mut event)]));
+ }
continue;
}
warn!("Otel Sink received log event when log support is disabled. Dropping trace");
@@ -170,13 +170,13 @@ impl Sink for OpenTelemetry {
error!("Failed to dispatch otel/gRPC logs message: {}", e);
self.is_down = true;
if event.transactional {
- return Ok(Some(vec![qos::close(&mut event)]));
- }
return Ok(Some(vec![
qos::fail(&mut event.clone()),
qos::close(&mut event),
]));
}
+ return Ok(Some(vec![qos::close(&mut event)]));
+ }
continue;
}
warn!("Otel Sink received trace event when trace support is disabled. Dropping trace");
@@ -189,7 +189,11 @@ impl Sink for OpenTelemetry {
}
self.is_down = false;
- return Ok(Some(vec![qos::ack(&mut event)]));
+ return Ok(Some(if event.transactional {
+ vec![qos::ack(&mut event)]
+ } else {
+ vec![]
+ }));
}
Ok(Some(vec![]))
