Ksql 13066 - Catch original GroupIdNotfoundException

ksqldb-engine/src/main/java/io/confluent/ksql/services/KafkaConsumerGroupClientImpl.java

@@ -34,12 +34,16 @@
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
 import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.apache.kafka.common.errors.RetriableException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 public class KafkaConsumerGroupClientImpl implements KafkaConsumerGroupClient {
 
   private final Supplier<Admin> adminClient;
+  private static final Logger log = LogManager.getLogger(KafkaConsumerGroupClientImpl.class);
 
   public KafkaConsumerGroupClientImpl(final Supplier<Admin> adminClient) {
     this.adminClient = adminClient;
@@ -83,6 +87,11 @@ public ConsumerGroupSummary describeConsumerGroup(final String group) {
       return new ConsumerGroupSummary(results);
     } catch (final GroupAuthorizationException e) {
       throw new KsqlGroupAuthorizationException(AclOperation.DESCRIBE, group);
+    } catch (final GroupIdNotFoundException e) {
+      // Per KIP-1043, Kafka now throws GroupIdNotFoundException if a group doesn't exist.
+      // Treat this expected scenario as a group with zero consumers by returning an empty summary.
+      log.debug("Consumer group '{}' not found, treating as empty. Reason: {}", group, e);
+      return new ConsumerGroupSummary(Collections.emptySet());
     } catch (final Exception e) {
       throw new KafkaResponseGetFailedException(
           "Failed to describe Kafka consumer groups: " + group, e);

ksqldb-engine/src/test/java/io/confluent/ksql/integration/KafkaConsumerGroupClientTest.java

@@ -104,16 +104,16 @@ public void shouldListConsumerGroupsWhenTheyExist() {
     verifyListsGroups(group1, ImmutableList.of(group0, group1));
   }
 
-//  @Test
-//  public void shouldDescribeConsumerGroup() {
-//    givenTopicExistsWithData();
-//    try (KafkaConsumer<String, byte[]> c1 = createConsumer(group0)) {
-//      verifyDescribeConsumerGroup(1, group0, ImmutableList.of(c1));
-//      try (KafkaConsumer<String, byte[]> c2 = createConsumer(group0)) {
-//        verifyDescribeConsumerGroup(2, group0, ImmutableList.of(c1, c2));
-//      }
-//    }
-//  }
+  @Test
+  public void shouldDescribeConsumerGroup() {
+    givenTopicExistsWithData();
+    try (KafkaConsumer<String, byte[]> c1 = createConsumer(group0)) {
+      verifyDescribeConsumerGroup(1, group0, ImmutableList.of(c1));
+      try (KafkaConsumer<String, byte[]> c2 = createConsumer(group0)) {
+        verifyDescribeConsumerGroup(2, group0, ImmutableList.of(c1, c2));
+      }
+    }
+  }
 
   @Test
   public void shouldListConsumerGroupOffsetsWhenTheyExist() {