bridge/kafka: Include partition in idempotency key

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

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 {

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.
     // <https://docs.rs/wiremock/latest/wiremock/struct.Mock.html#method.expect>
     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": {