KafkaSinkCluster routing fixes

shotover · Nov 14, 2024 · 21143af · 21143af
1 parent 0310cbf
commit 21143af
Showing 1 changed file with 14 additions and 0 deletions.
diff --git a/shotover/src/transforms/kafka/sink_cluster/mod.rs b/shotover/src/transforms/kafka/sink_cluster/mod.rs
@@ -697,6 +697,14 @@ impl KafkaSinkCluster {
                         self.store_topic_names(&mut topic_names, topic.topic.clone());
                     }
                 }
+                Some(Frame::Kafka(KafkaFrame::Request {
+                    body: RequestBody::DeleteRecords(body),
+                    ..
+                })) => {
+                    for topic in &body.topics {
+                        self.store_topic_names(&mut topic_names, topic.name.clone());
+                    }
+                }
                 Some(Frame::Kafka(KafkaFrame::Request {
                     body: RequestBody::Fetch(fetch),
                     ..
@@ -728,6 +736,12 @@ impl KafkaSinkCluster {
                         self.store_group(&mut groups, group_id.clone());
                     }
                 }
+                Some(Frame::Kafka(KafkaFrame::Request {
+                    body: RequestBody::OffsetDelete(offset_delete),
+                    ..
+                })) => {
+                    self.store_group(&mut groups, offset_delete.group_id.clone());
+                }
                 Some(Frame::Kafka(KafkaFrame::Request {
                     body:
                         RequestBody::InitProducerId(InitProducerIdRequest {