feat: add consumerGroupId to QueryDescription (#8073)

* feat: add consumerGroupId to QueryDescription

ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/ClientTest.java

@@ -1021,7 +1021,7 @@ public void shouldFailToExplainQueryViaExecuteStatement() {
         new QueryDescription(new QueryId("id"), "sql", Optional.empty(),
             Collections.emptyList(), Collections.emptySet(), Collections.emptySet(), "topology",
             "executionPlan", Collections.emptyMap(), Collections.emptyMap(),
-            KsqlQueryType.PERSISTENT, Collections.emptyList(), Collections.emptySet()));
+            KsqlQueryType.PERSISTENT, Collections.emptyList(), Collections.emptySet(), "consumerGroupId"));
     testEndpoints.setKsqlEndpointResponse(Collections.singletonList(entity));
 
     // When

ksqldb-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java

@@ -372,7 +372,8 @@ public void shouldPrintExplainQueryWithError() {
                     Collections.emptySet(),
                     Optional.empty()
                 )
-            )
+            ),
+            "consumerGroupId"
         )
     );
 

ksqldb-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.shouldPrintExplainQueryWithError.approved.json

@@ -34,6 +34,7 @@
       "topicOffsets" : [ ],
       "timeCurrentIdlingStarted" : null
     } ],
+    "consumerGroupId" : "consumerGroupId",
     "state" : "ERROR"
   },
   "warnings" : [ ]

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/entity/QueryDescriptionFactory.java

@@ -75,7 +75,8 @@ private static QueryDescription create(
         ksqlHostQueryStatus,
         queryMetadata.getQueryType(),
         queryMetadata.getQueryErrors(),
-        queryMetadata.getTaskMetadata()
+        queryMetadata.getTaskMetadata(),
+        queryMetadata.getQueryApplicationId()
     );
   }
 }

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/execution/ExplainExecutorTest.java

@@ -179,6 +179,7 @@ public void shouldFailOnNonQueryExplain() {
   @SuppressWarnings("SameParameterValue")
   public static PersistentQueryMetadata givenPersistentQuery(final String id) {
     final PersistentQueryMetadata metadata = mock(PersistentQueryMetadata.class);
+    when(metadata.getQueryApplicationId()).thenReturn("consumer-group-id");
     when(metadata.getQueryId()).thenReturn(new QueryId(id));
     when(metadata.getSinkName()).thenReturn(Optional.of(SourceName.of(id)));
     when(metadata.getLogicalSchema()).thenReturn(TemporaryEngine.SCHEMA);

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/execution/ListQueriesExecutorTest.java

@@ -500,6 +500,7 @@ public static void mockQuery(
     when(metadata.getState()).thenReturn(state);
     when(metadata.getTopologyDescription()).thenReturn("topology");
     when(metadata.getExecutionPlan()).thenReturn("plan");
+    when(metadata.getQueryApplicationId()).thenReturn("consumer-group-id");
 
     final StreamsMetadata localStreamsMetadata = mock(StreamsMetadata.class);
     when(localStreamsMetadata.hostInfo()).thenReturn(LOCAL_HOST);

ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/entity/QueryDescription.java

@@ -53,6 +53,7 @@ public class QueryDescription {
   private final KsqlQueryType queryType;
   private final List<QueryError> queryErrors;
   private final Set<StreamsTaskMetadata> tasksMetadata;
+  private final String consumerGroupId;
 
   // CHECKSTYLE_RULES.OFF: ParameterNumberCheck
   @SuppressWarnings("WeakerAccess") // Invoked via reflection
@@ -71,7 +72,8 @@ public QueryDescription(
           ksqlHostQueryStatus,
       @JsonProperty("queryType") final KsqlQueryType queryType,
       @JsonProperty("queryErrors") final List<QueryError> queryErrors,
-      @JsonProperty("tasksMetadata") final Set<StreamsTaskMetadata> tasksMetadata
+      @JsonProperty("tasksMetadata") final Set<StreamsTaskMetadata> tasksMetadata,
+      @JsonProperty("consumerGroupId") final String getConsumerGroupId
   ) {
     this.id = Objects.requireNonNull(id, "id");
     this.statementText = Objects.requireNonNull(statementText, "statementText");
@@ -88,12 +90,17 @@ public QueryDescription(
     this.queryType = Objects.requireNonNull(queryType, "queryType");
     this.queryErrors = new ArrayList<>(Objects.requireNonNull(queryErrors, "queryErrors"));
     this.tasksMetadata = new HashSet<>(Objects.requireNonNull(tasksMetadata));
+    this.consumerGroupId = Objects.requireNonNull(getConsumerGroupId, "consumerGroupId");
   }
 
   public QueryId getId() {
     return id;
   }
 
+  public String getConsumerGroupId() {
+    return consumerGroupId;
+  }
+
   public String getStatementText() {
     return statementText;
   }
@@ -195,7 +202,8 @@ public boolean equals(final Object o) {
         && Objects.equals(ksqlHostQueryStatus, that.ksqlHostQueryStatus)
         && Objects.equals(queryType, that.queryType)
         && Objects.equals(queryErrors, that.queryErrors)
-        && Objects.equals(tasksMetadata, that.tasksMetadata);
+        && Objects.equals(tasksMetadata, that.tasksMetadata)
+        && Objects.equals(consumerGroupId, that.consumerGroupId);
   }
 
   @Override
@@ -213,7 +221,8 @@ public int hashCode() {
         ksqlHostQueryStatus,
         queryType,
         queryErrors,
-        tasksMetadata
+        tasksMetadata,
+        consumerGroupId
     );
   }
 }