Skip to content

Commit

Permalink
fix(kafka sink): retry messages that result in kafka policy violations
Browse files Browse the repository at this point in the history
Problem: Some messages were getting dropped by Vector due to Kafka
throwing `PolicyViolation` errors. These should be retried as a policy
can be as simple as a more aggressive rate limit.

----------

Solution: Retry any messages that had the `RDKafkaErrorCode::PolicyViolation` error.

----------

Note: A dynamic back off may be better, as there may be a rate limit out
there that still needs more than 100ms to back off on requests.

----------

See the original issue at #22026

Closes #22026
  • Loading branch information
PriceHiller committed Dec 16, 2024
1 parent 7e93489 commit 296551f
Show file tree
Hide file tree
Showing 2 changed files with 8 additions and 2 deletions.
3 changes: 3 additions & 0 deletions changelog.d/22026-retry-kafka-policy-violations.fix.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
Retry Kafka messages that error with `RDKafkaErrorCode::PolicyViolation` so messages are not lost.

authors: PriceHiller
7 changes: 5 additions & 2 deletions src/sinks/kafka/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -159,9 +159,12 @@ impl Service<KafkaRequest> for KafkaService {
})
.map_err(|(err, _)| err);
}
// Producer queue is full.
// Producer queue is full or a policy has been violated and the request should
// be retried
Err((
KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull),
KafkaError::MessageProduction(
RDKafkaErrorCode::QueueFull | RDKafkaErrorCode::PolicyViolation,
),
original_record,
)) => {
if blocked_state.is_none() {
Expand Down

0 comments on commit 296551f

Please sign in to comment.