From 3b94dab3ce53a02f35e7b278ce0a2506c78c0454 Mon Sep 17 00:00:00 2001 From: shivanshu Date: Mon, 26 Aug 2024 15:36:18 +0530 Subject: [PATCH] chore: small patch to fix consumer_group check --- .../app/integrations/messagingQueues/kafka/translator.go | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/pkg/query-service/app/integrations/messagingQueues/kafka/translator.go b/pkg/query-service/app/integrations/messagingQueues/kafka/translator.go index 98414ebf0f..6b5d355caa 100644 --- a/pkg/query-service/app/integrations/messagingQueues/kafka/translator.go +++ b/pkg/query-service/app/integrations/messagingQueues/kafka/translator.go @@ -48,15 +48,14 @@ func buildClickHouseQuery(messagingQueue *MessagingQueue, queueType string, quer return nil, fmt.Errorf("invalid type for Partition") } - consumerGroup, ok := messagingQueue.Variables["consumer_group"] - if !ok { - return nil, fmt.Errorf("invalid type for consumer group") - } - var query string if queryContext == "producer" { query = generateProducerSQL(start, end, topic, partition, queueType) } else if queryContext == "consumer" { + consumerGroup, ok := messagingQueue.Variables["consumer_group"] + if !ok { + return nil, fmt.Errorf("invalid type for consumer group") + } query = generateConsumerSQL(start, end, topic, partition, consumerGroup, queueType) }