feat: Add SHOW TOPICS EXTENDED

Fixes #1268 BREAKING CHANGE: "SHOW TOPICS" no longer includes the "Consumers" and "ConsumerGroups" columns. You can use "SHOW TOPICS EXTENDED" to get the output previous emitted from "SHOW TOPICS". See below for examples. This change splits "SHOW TOPICS" into two commands: 1. "SHOW TOPICS EXTENDED", which shows what was previously shown by "SHOW TOPICS". Sample output: ``` ksql> show topics extended; Kafka Topic | Partitions | Partition Replicas | Consumers | ConsumerGroups -------------------------------------------------------------------------------------------------------------------------------------------------------------- _confluent-command | 1 | 1 | 1 | 1 _confluent-controlcenter-5-3-0-1-actual-group-consumption-rekey | 1 | 1 | 1 | 1 ``` 2. "SHOW TOPICS", which now no longer queries consumer groups and their active consumers. Sample output: ``` ksql> show topics; Kafka Topic | Partitions | Partition Replicas --------------------------------------------------------------------------------------------------------------------------------- _confluent-command | 1 | 1 _confluent-controlcenter-5-3-0-1-actual-group-consumption-rekey | 1 | 1 ``` This changeset primarily involved renaming KafkaTopicXXX classes to KafkaTopicXXXExtended and then trimming out the consumer group info from the original KafkaTopicXXX classes.
confluentinc · Aug 7, 2019 · a6fe3b7 · a6fe3b7
1 parent b503c8d
commit a6fe3b7
Show file tree

Hide file tree

Showing 21 changed files with 501 additions and 236 deletions.
diff --git a/docs/developer-guide/query-with-structured-data.rst b/docs/developer-guide/query-with-structured-data.rst
@@ -121,12 +121,12 @@ Your output should resemble:
 
 ::
 
-     Kafka Topic        | Partitions | Partition Replicas | Consumers | ConsumerGroups
-    ----------------------------------------------------------------------------------
-     _confluent-metrics | 12         | 1                  | 0         | 0
-     _schemas           | 1          | 1                  | 0         | 0
-     raw-topic          | 1          | 1                  | 0         | 0
-    ----------------------------------------------------------------------------------
+     Kafka Topic        | Partitions | Partition Replicas
+    ------------------------------------------------------
+     _confluent-metrics | 12         | 1
+     _schemas           | 1          | 1
+     raw-topic          | 1          | 1
+    ------------------------------------------------------
 
 Inspect ``raw-topic`` to ensure that |kcat| populated it: 
 

diff --git a/docs/developer-guide/syntax-reference.rst b/docs/developer-guide/syntax-reference.rst
@@ -1338,13 +1338,14 @@ SHOW TOPICS
 
 .. code:: sql
 
-    SHOW | LIST TOPICS;
+    SHOW | LIST TOPICS [EXTENDED];
 
 **Description**
 
-List the available topics in the Kafka cluster that KSQL is configured
+SHOW TOPICS lists the available topics in the Kafka cluster that KSQL is configured
 to connect to (default setting for ``bootstrap.servers``:
-``localhost:9092``).
+``localhost:9092``). SHOW TOPICS EXTENDED also displays consumer groups and their active consumer
+counts.
 
 .. _show-streams:
 

diff --git a/docs/includes/ksql-includes.rst b/docs/includes/ksql-includes.rst
@@ -126,13 +126,13 @@ Your output should resemble:
 
 ::
 
-     Kafka Topic        | Partitions | Partition Replicas | Consumers | ConsumerGroups
-    -----------------------------------------------------------------------------------
-     _confluent-metrics | 12         | 1                  | 0         | 0
-     _schemas           | 1          | 1                  | 0         | 0
-     pageviews          | 1          | 1                  | 0         | 0
-     users              | 1          | 1                  | 0         | 0
-    -----------------------------------------------------------------------------------
+     Kafka Topic        | Partitions | Partition Replicas
+    ------------------------------------------------------
+     _confluent-metrics | 12         | 1
+     _schemas           | 1          | 1
+     pageviews          | 1          | 1
+     users              | 1          | 1
+    ------------------------------------------------------
 
 Inspect the ``users`` topic by using the PRINT statement:
 

diff --git a/docs/tutorials/basics-control-center.rst b/docs/tutorials/basics-control-center.rst
@@ -154,9 +154,7 @@ statements in KSQL Editor, just like you use them in the KSQL CLI.
          "registered": true,
          "replicaInfo": [
            1
-         ],
-         "consumerCount": 0,
-         "consumerGroupCount": 0
+         ]
        },
 
    The ``"registered": true`` indicator means that you have registered the topic

diff --git a/ksql-cli/src/main/java/io/confluent/ksql/cli/console/Console.java b/ksql-cli/src/main/java/io/confluent/ksql/cli/console/Console.java
@@ -49,6 +49,7 @@
 import io.confluent.ksql.rest.entity.FunctionInfo;
 import io.confluent.ksql.rest.entity.FunctionNameList;
 import io.confluent.ksql.rest.entity.KafkaTopicsList;
+import io.confluent.ksql.rest.entity.KafkaTopicsListExtended;
 import io.confluent.ksql.rest.entity.KsqlEntity;
 import io.confluent.ksql.rest.entity.KsqlEntityList;
 import io.confluent.ksql.rest.entity.KsqlErrorMessage;
@@ -130,7 +131,11 @@ public class Console implements Closeable {
           .put(TablesList.class,
               tablePrinter(TablesList.class, TablesListTableBuilder::new))
           .put(KafkaTopicsList.class,
-              tablePrinter(KafkaTopicsList.class, KafkaTopicsListTableBuilder::new))
+              tablePrinter(KafkaTopicsList.class, KafkaTopicsListTableBuilder.SimpleBuilder::new))
+          .put(KafkaTopicsListExtended.class,
+              tablePrinter(
+                  KafkaTopicsListExtended.class,
+                  KafkaTopicsListTableBuilder.ExtendedBuilder::new))
           .put(ExecutionPlan.class,
               tablePrinter(ExecutionPlan.class, ExecutionPlanTableBuilder::new))
           .put(FunctionNameList.class,

diff --git a/...rc/main/java/io/confluent/ksql/cli/console/table/builder/KafkaTopicsListTableBuilder.java b/...rc/main/java/io/confluent/ksql/cli/console/table/builder/KafkaTopicsListTableBuilder.java
@@ -19,33 +19,57 @@
 import io.confluent.ksql.cli.console.table.Table;
 import io.confluent.ksql.cli.console.table.Table.Builder;
 import io.confluent.ksql.rest.entity.KafkaTopicsList;
+import io.confluent.ksql.rest.entity.KafkaTopicsListExtended;
 import io.confluent.ksql.util.StringUtil;
 import java.util.List;
 import java.util.stream.Stream;
 
-public class KafkaTopicsListTableBuilder implements TableBuilder<KafkaTopicsList> {
+public class KafkaTopicsListTableBuilder {
 
-  private static final List<String> HEADERS = ImmutableList.of(
-      "Kafka Topic",
-      "Partitions",
-      "Partition Replicas",
-      "Consumers",
-      "ConsumerGroups");
+  public static class SimpleBuilder implements TableBuilder<KafkaTopicsList> {
+    private static final List<String> HEADERS = ImmutableList.of(
+        "Kafka Topic",
+        "Partitions",
+        "Partition Replicas");
 
-  @Override
-  public Table buildTable(final KafkaTopicsList entity) {
-    final Stream<List<String>> rows = entity.getTopics().stream()
-        .map(t -> ImmutableList.of(
-            t.getName(),
-            Integer.toString(t.getReplicaInfo().size()),
-            getTopicReplicaInfo(t.getReplicaInfo()),
-            Integer.toString(t.getConsumerCount()),
-            Integer.toString(t.getConsumerGroupCount())));
+    @Override
+    public Table buildTable(final KafkaTopicsList entity) {
+      final Stream<List<String>> rows = entity.getTopics().stream()
+          .map(t -> ImmutableList.of(
+              t.getName(),
+              Integer.toString(t.getReplicaInfo().size()),
+              getTopicReplicaInfo(t.getReplicaInfo())));
 
-    return new Builder()
-        .withColumnHeaders(HEADERS)
-        .withRows(rows)
-        .build();
+      return new Builder()
+          .withColumnHeaders(HEADERS)
+          .withRows(rows)
+          .build();
+    }
+  }
+
+  public static class ExtendedBuilder implements TableBuilder<KafkaTopicsListExtended> {
+    private static final List<String> HEADERS = ImmutableList.of(
+        "Kafka Topic",
+        "Partitions",
+        "Partition Replicas",
+        "Consumers",
+        "ConsumerGroups");
+
+    @Override
+    public Table buildTable(final KafkaTopicsListExtended entity) {
+      final Stream<List<String>> rows = entity.getTopics().stream()
+          .map(t -> ImmutableList.of(
+              t.getName(),
+              Integer.toString(t.getReplicaInfo().size()),
+              getTopicReplicaInfo(t.getReplicaInfo()),
+              Integer.toString(t.getConsumerCount()),
+              Integer.toString(t.getConsumerGroupCount())));
+
+      return new Builder()
+          .withColumnHeaders(HEADERS)
+          .withRows(rows)
+          .build();
+    }
   }
 
   /**

diff --git a/ksql-cli/src/test/java/io/confluent/ksql/cli/CliTest.java b/ksql-cli/src/test/java/io/confluent/ksql/cli/CliTest.java
@@ -355,9 +355,17 @@ private void selectWithLimit(
 
   @Test
   public void shouldPrintResultsForListOrShowCommands() {
-
     assertRunListCommand(
         "topics",
+        hasRow(
+            equalTo(orderDataProvider.topicName()),
+            equalTo("1"),
+            equalTo("1")
+        )
+    );
+
+    assertRunListCommand(
+        "topics extended",
         hasRow(
             equalTo(orderDataProvider.topicName()),
             equalTo("1"),

diff --git a/ksql-parser/src/main/antlr4/io/confluent/ksql/parser/SqlBase.g4 b/ksql-parser/src/main/antlr4/io/confluent/ksql/parser/SqlBase.g4
@@ -34,7 +34,7 @@ singleExpression
 statement
     : query                                                                 #querystatement
     | (LIST | SHOW) PROPERTIES                                              #listProperties
-    | (LIST | SHOW) TOPICS                                                  #listTopics
+    | (LIST | SHOW) TOPICS EXTENDED?                                        #listTopics
     | (LIST | SHOW) STREAMS EXTENDED?                                       #listStreams
     | (LIST | SHOW) TABLES EXTENDED?                                        #listTables
     | (LIST | SHOW) FUNCTIONS                                               #listFunctions

diff --git a/ksql-parser/src/main/java/io/confluent/ksql/parser/AstBuilder.java b/ksql-parser/src/main/java/io/confluent/ksql/parser/AstBuilder.java
@@ -571,7 +571,7 @@ public Node visitRunScript(final SqlBaseParser.RunScriptContext context) {
 
     @Override
     public Node visitListTopics(final SqlBaseParser.ListTopicsContext context) {
-      return new ListTopics(getLocation(context));
+      return new ListTopics(getLocation(context), context.EXTENDED() != null);
     }
 
     @Override

diff --git a/ksql-parser/src/main/java/io/confluent/ksql/parser/tree/ListTopics.java b/ksql-parser/src/main/java/io/confluent/ksql/parser/tree/ListTopics.java
@@ -22,27 +22,38 @@
 
 public class ListTopics extends Statement {
 
-  public ListTopics(final Optional<NodeLocation> location) {
+  private final boolean showExtended;
+
+  public ListTopics(final Optional<NodeLocation> location, final boolean showExtended) {
     super(location);
+    this.showExtended = showExtended;
   }
 
-  @Override
-  public int hashCode() {
-    return Objects.hash(getClass());
+  public boolean getShowExtended() {
+    return showExtended;
   }
 
   @Override
-  public boolean equals(final Object obj) {
-    if (this == obj) {
+  public boolean equals(final Object o) {
+    if (this == o) {
       return true;
     }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final ListTopics that = (ListTopics) o;
+    return showExtended == that.showExtended;
+  }
 
-    return obj != null && obj.getClass().equals(getClass());
+  @Override
+  public int hashCode() {
+    return Objects.hash(showExtended);
   }
 
   @Override
   public String toString() {
     return toStringHelper(this)
+        .add("showExtended", showExtended)
         .toString();
   }
 }
diff --git a/ksql-parser/src/test/java/io/confluent/ksql/parser/KsqlParserTest.java b/ksql-parser/src/test/java/io/confluent/ksql/parser/KsqlParserTest.java
@@ -619,11 +619,17 @@ public void testSelectSessionWindow() {
 
   @Test
   public void testShowTopics() {
+    // Given:
     final String simpleQuery = "SHOW TOPICS;";
+
+    // When:
     final Statement statement = KsqlParserTestUtil.buildSingleAst(simpleQuery, metaStore).getStatement();
-    Assert.assertTrue(statement instanceof ListTopics);
     final ListTopics listTopics = (ListTopics) statement;
-    Assert.assertTrue(listTopics.toString().equalsIgnoreCase("ListTopics{}"));
+
+    // Then:
+    Assert.assertTrue(statement instanceof ListTopics);
+    Assert.assertThat(listTopics.toString(), is("ListTopics{showExtended=false}"));
+    Assert.assertThat(listTopics.getShowExtended(), is(false));
   }
 
   @Test
@@ -761,6 +767,21 @@ public void shouldSetShowDescriptionsForShowStreamsDescriptions() {
     Assert.assertThat(listStreams.getShowExtended(), is(true));
   }
 
+  @Test
+  public void shouldSetShowDescriptionsForShowTopicsDescriptions() {
+    // Given:
+    final String statementString = "SHOW TOPICS EXTENDED;";
+
+    // When:
+    final Statement statement = KsqlParserTestUtil.buildSingleAst(statementString, metaStore)
+        .getStatement();
+
+    // Then:
+    Assert.assertThat(statement, instanceOf(ListTopics.class));
+    final ListTopics listTopics = (ListTopics)statement;
+    Assert.assertThat(listTopics.getShowExtended(), is(true));
+  }
+
   @Test
   public void shouldSetShowDescriptionsForShowTablesDescriptions() {
     final String statementString = "SHOW TABLES EXTENDED;";

diff --git a/ksql-parser/src/test/java/io/confluent/ksql/parser/tree/ListTopicsTest.java b/ksql-parser/src/test/java/io/confluent/ksql/parser/tree/ListTopicsTest.java
@@ -26,11 +26,15 @@ public class ListTopicsTest {
 
   @Test
   public void shouldImplementHashCodeAndEqualsProperty() {
+    // Note: At the moment location does not take part in equality testing
     new EqualsTester()
         .addEqualityGroup(
-            // Note: At the moment location does not take part in equality testing
-            new ListTopics(Optional.of(SOME_LOCATION)),
-            new ListTopics(Optional.of(OTHER_LOCATION))
+            new ListTopics(Optional.of(SOME_LOCATION), true),
+            new ListTopics(Optional.of(OTHER_LOCATION), true)
+        )
+        .addEqualityGroup(
+            new ListTopics(Optional.of(SOME_LOCATION), false),
+            new ListTopics(Optional.of(OTHER_LOCATION), false)
         )
         .testEquals();
   }

diff --git a/ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/KafkaTopicInfo.java b/ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/KafkaTopicInfo.java
@@ -28,20 +28,14 @@ public class KafkaTopicInfo {
 
   private final String name;
   private final List<Integer> replicaInfo;
-  private final int consumerGroupCount;
-  private final int consumerCount;
 
   @JsonCreator
   public KafkaTopicInfo(
       @JsonProperty("name") final String name,
-      @JsonProperty("replicaInfo") final List<Integer> replicaInfo,
-      @JsonProperty("consumerCount") final int consumerCount,
-      @JsonProperty("consumerGroupCount") final int consumerGroupCount
+      @JsonProperty("replicaInfo") final List<Integer> replicaInfo
   ) {
     this.name = name;
     this.replicaInfo = replicaInfo;
-    this.consumerGroupCount = consumerGroupCount;
-    this.consumerCount = consumerCount;
   }
 
   public String getName() {
@@ -52,14 +46,6 @@ public List<Integer> getReplicaInfo() {
     return replicaInfo;
   }
 
-  public int getConsumerCount() {
-    return consumerCount;
-  }
-
-  public int getConsumerGroupCount() {
-    return consumerGroupCount;
-  }
-
   @Override
   public boolean equals(final Object o) {
     if (this == o) {