diff --git a/ChangeLog.md b/ChangeLog.md index f3c60d3beb..fe2b3ef341 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -2,6 +2,7 @@ ## Unreleased * Server: Fix warning when opentelemetry is disabled +* Bridge: Include the Kafka partition in consumer idempotency keys ## Version 1.98.0 * CLI, Server, Bridge: Set OCI metadata on Docker images diff --git a/bridge/svix-bridge-plugin-kafka/src/input.rs b/bridge/svix-bridge-plugin-kafka/src/input.rs index 8bcaccabc4..f80f37cdc6 100644 --- a/bridge/svix-bridge-plugin-kafka/src/input.rs +++ b/bridge/svix-bridge-plugin-kafka/src/input.rs @@ -81,9 +81,11 @@ impl KafkaConsumer { // If committing the message fails or the process crashes after posting the webhook but // before committing, this makes sure that the next run of this fn with the same kafka // message doesn't end up creating a duplicate webhook in svix. - idempotency_key: Some(format!( - "svix_bridge_kafka_{group_id}_{topic}_{}", - msg.offset() + idempotency_key: Some(kafka_idempotency_key( + group_id, + topic, + msg.partition(), + msg.offset(), )), }; @@ -183,6 +185,10 @@ impl KafkaConsumer { } } +fn kafka_idempotency_key(group_id: &str, topic: &str, partition: i32, offset: i64) -> String { + format!("svix_bridge_kafka_{group_id}_{topic}_{partition}_{offset}") +} + #[async_trait] impl SenderInput for KafkaConsumer { fn name(&self) -> &str { diff --git a/bridge/svix-bridge-plugin-kafka/tests/it/kafka_consumer.rs b/bridge/svix-bridge-plugin-kafka/tests/it/kafka_consumer.rs index 00454c7a98..3622f10e0b 100644 --- a/bridge/svix-bridge-plugin-kafka/tests/it/kafka_consumer.rs +++ b/bridge/svix-bridge-plugin-kafka/tests/it/kafka_consumer.rs @@ -19,7 +19,7 @@ use svix_bridge_types::{ use tracing::info; use wiremock::{ Mock, MockServer, ResponseTemplate, - matchers::{body_partial_json, method}, + matchers::{body_partial_json, header, method}, }; use crate::{BROKER_HOST, create_topic, delete_topic, kafka_admin_client}; @@ -100,6 +100,8 @@ async fn publish(producer: &FutureProducer, topic: &str, payload: &[u8]) { #[tokio::test] async fn test_consume_ok() { let topic = unique_topic_name!(); + let expected_idempotency_key = + format!("svix_bridge_kafka_svix_bridge_test_group_id_{topic}_0_0"); let admin_client = kafka_admin_client(); create_topic(&admin_client, topic).await; @@ -111,6 +113,7 @@ async fn test_consume_ok() { // The `expect` call should ensure we see exactly 1 POST request. // let mock = Mock::given(method("POST")) + .and(header("idempotency-key", expected_idempotency_key)) .respond_with(ResponseTemplate::new(202).set_body_json(json!({ "eventType": "testing.things", "payload": {