Messages getting dropped unintentionally on kafka sink

[heshanperera-alert]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

I am using Kafka sink to send messages to azure eventhub. I have noted that messages gets dropped when the azure event-hub hits the throttling limits. When I checked the vector's internal metrics, i can see that these messages gets dropped unintenionally. Is there a way to determine whether these events are retried automatically by vector?

I am going through the documentation on Kafka sink, but there doesn't seem to have a way or a flag to retry these events upon failures

Version

0.37.1

Debug Output

vector_common::internal_event::service: Service call failed. No retries or retries exhausted. error=Some(KafkaError (Message production error: PolicyViolation (Broker: Policy violation)))

[jszwedko]

Thanks @heshanperera-alert . It does seem like this error should be retried. I think the change would need to be somewhere in here if anyone feels motivated:

vector/src/sinks/kafka/service.rs

Lines 147 to 199 in f73fb10

	match this.kafka_producer.send_result(record) {
	// Record was successfully enqueued on the producer.
	Ok(fut) => {
	// Drop the blocked state (if any), as the producer is no longer blocked.
	drop(blocked_state.take());
	return fut
	.await
	.expect("producer unexpectedly dropped")
	.map(|_| KafkaResponse {
	event_byte_size,
	raw_byte_size,
	event_status: EventStatus::Delivered,
	})
	.map_err(|(err, _)| err);
	}
	// Producer queue is full.
	Err((
	KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull),
	original_record,
	)) => {
	if blocked_state.is_none() {
	blocked_state =
	Some(BlockedRecordState::new(Arc::clone(&this.records_blocked)));
	}
	record = original_record;
	tokio::time::sleep(Duration::from_millis(100)).await;
	}
	// A final/non-retriable error occurred.
	Err((
	err @ KafkaError::MessageProduction(
	RDKafkaErrorCode::InvalidMessage
	| RDKafkaErrorCode::InvalidMessageSize
	| RDKafkaErrorCode::MessageSizeTooLarge
	| RDKafkaErrorCode::UnknownTopicOrPartition
	| RDKafkaErrorCode::InvalidRecord
	| RDKafkaErrorCode::InvalidRequiredAcks
	| RDKafkaErrorCode::TopicAuthorizationFailed
	| RDKafkaErrorCode::UnsupportedForMessageFormat
	| RDKafkaErrorCode::ClusterAuthorizationFailed,
	),
	_,
	)) => return Err(err),
	// A different error occurred. Set event status to Errored not Rejected.
	Err(_) => {
	return Ok(KafkaResponse {
	event_byte_size: config::telemetry().create_request_count_byte_size(),
	raw_byte_size: 0,
	event_status: EventStatus::Errored,
	})
	}
	};
	}

[PriceHiller]

So would a simple fix be to shove the RDKafkaErrorCode::PolicyVioltion into that area like so?

diff --git a/src/sinks/kafka/service.rs b/src/sinks/kafka/service.rs
index 087795864..4772833e9 100644
--- a/src/sinks/kafka/service.rs
+++ b/src/sinks/kafka/service.rs
@@ -161,7 +161,7 @@ impl Service<KafkaRequest> for KafkaService {
                     }
                     // Producer queue is full.
                     Err((
-                        KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull),
+                        KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull | RDKafkaErrorCode::PolicyViolation),
                         original_record,
                     )) => {
                         if blocked_state.is_none() {

I'm not entirely certain how that chunk of code works, it appears to just delay by a bit and retry the request after 100ms if I'm understanding it right.

[jszwedko]

So would a simple fix be to shove the RDKafkaErrorCode::PolicyVioltion into that area like so?

diff --git a/src/sinks/kafka/service.rs b/src/sinks/kafka/service.rs
index 087795864..4772833e9 100644
--- a/src/sinks/kafka/service.rs
+++ b/src/sinks/kafka/service.rs
@@ -161,7 +161,7 @@ impl Service<KafkaRequest> for KafkaService {
                     }
                     // Producer queue is full.
                     Err((
-                        KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull),
+                        KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull | RDKafkaErrorCode::PolicyViolation),
                         original_record,
                     )) => {
                         if blocked_state.is_none() {

I'm not entirely certain how that chunk of code works, it appears to just delay by a bit and retry the request after 100ms if I'm understanding it right.

Yeah, I think that would work. And yes, it just waits 100ms and retries.