Fix some more rest api inconsistencies (#1210)

Fix some more rest api inconsistencies

- Use the name sourceDescription(s) for fields that contain a
SourceDescription or List<SourceDescription>
- Use the name queryDescription(s) for fields that contain a
QueryDescription or List<QueryDescription>
- List sinks in the RunningQuery object (instead of a topic name)
- Return a simple string for query ids
- Include sinks in QueryDescriptions. Previously we had a bug where
they were not being listed.
- Display read queries in SourceDescriptions. Previously we had a bug
where readQueries actually listed queries writing into a source
- Return RunningQuery objects in SourceDescription

Fixes #1199
Fixes #1206

Author: rodesai

ksql-cli/src/main/java/io/confluent/ksql/cli/console/Console.java

@@ -25,6 +25,7 @@
 import io.confluent.ksql.rest.entity.QueryDescription;
 import io.confluent.ksql.rest.entity.QueryDescriptionEntity;
 import io.confluent.ksql.rest.entity.QueryDescriptionList;
+import io.confluent.ksql.rest.entity.RunningQuery;
 import io.confluent.ksql.rest.entity.SourceDescriptionEntity;
 import io.confluent.ksql.rest.entity.SourceDescriptionList;
 import org.apache.commons.lang3.StringUtils;
@@ -385,10 +386,11 @@ private void printAsTable(KsqlEntity ksqlEntity) {
           .withColumnHeaders(PROPERTIES_COLUMN_HEADERS)
           .withRows(propertiesRowValues(properties));
     } else if (ksqlEntity instanceof Queries) {
-      List<Queries.RunningQuery> runningQueries = ((Queries) ksqlEntity).getQueries();
+      List<RunningQuery> runningQueries = ((Queries) ksqlEntity).getQueries();
       tableBuilder.withColumnHeaders("Query ID", "Kafka Topic", "Query String");
       runningQueries.forEach(
-          r -> tableBuilder.withRow(r.getId().toString(), r.getKafkaTopic(), r.getQueryString()));
+          r -> tableBuilder.withRow(
+              r.getId(), String.join(",", r.getSinks()), r.getQueryString()));
       tableBuilder.withFooterLine("For detailed information on a Query run: EXPLAIN <Query ID>;");
     } else if (ksqlEntity instanceof SourceDescriptionEntity) {
       SourceDescriptionEntity sourceDescriptionEntity = (SourceDescriptionEntity) ksqlEntity;
@@ -502,7 +504,7 @@ private void printTopicInfo(SourceDescription source) {
     writer().println(String.format("%-20s : %s", "Key field", source.getKey()));
     writer().println(String.format("%-20s : %s", "Key format", "STRING"));
     writer().println(String.format("%-20s : %s", "Timestamp field", timestamp));
-    writer().println(String.format("%-20s : %s", "Value format", source.getSerdes()));
+    writer().println(String.format("%-20s : %s", "Value format", source.getFormat()));
 
     if (!source.getTopic().isEmpty()) {
       writer().println(String.format(
@@ -522,8 +524,8 @@ private void printWriteQueries(SourceDescription source) {
           "Queries that write into this " + source.getType(),
           "-----------------------------------"
       ));
-      for (String writeQuery : source.getWriteQueries()) {
-        writer().println(writeQuery);
+      for (RunningQuery writeQuery : source.getWriteQueries()) {
+        writer().println(writeQuery.getId() + " : " + writeQuery.getQueryString());
       }
       writer().println("\nFor query topology and execution plan please run: EXPLAIN <QueryId>");
     }
@@ -633,7 +635,7 @@ private void printQuerySinks(QueryDescription query) {
   }
 
   private void printQueryDescription(QueryDescription query) {
-    writer().println(String.format("%-20s : %s", "ID", query.getId().getId()));
+    writer().println(String.format("%-20s : %s", "ID", query.getId()));
     if (query.getStatementText().length() > 0) {
       writer().println(String.format("%-20s : %s", "SQL", query.getStatementText()));
     }

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

@@ -18,6 +18,7 @@
 
 import com.google.common.collect.ImmutableList;
 
+import io.confluent.ksql.rest.entity.RunningQuery;
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.junit.After;
@@ -36,7 +37,6 @@
 import io.confluent.ksql.FakeException;
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.TestTerminal;
-import io.confluent.ksql.query.QueryId;
 import io.confluent.ksql.rest.client.KsqlRestClient;
 import io.confluent.ksql.rest.entity.CommandStatusEntity;
 import io.confluent.ksql.rest.entity.ExecutionPlan;
@@ -102,8 +102,8 @@ public void testPrintKSqlEntityList() throws IOException {
     properties.put("k2", "v2");
     properties.put("k3", true);
 
-    List<Queries.RunningQuery> queries = new ArrayList<>();
-    queries.add(new Queries.RunningQuery("select * from t1", "TestTopic", new QueryId("0")));
+    List<RunningQuery> queries = new ArrayList<>();
+    queries.add(new RunningQuery("select * from t1", Collections.singleton("Test"), "0"));
 
     for (int i = 0; i < 5; i++) {
       KsqlEntityList entityList = new KsqlEntityList(ImmutableList.of(

ksql-common/src/main/java/io/confluent/ksql/query/QueryId.java

@@ -17,11 +17,9 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonSubTypes;
 
 import java.util.Objects;
 
-@JsonSubTypes({})
 public class QueryId {
   private final String id;
 

ksql-engine/src/main/java/io/confluent/ksql/KsqlContext.java

@@ -90,8 +90,6 @@ public void sql(String sql) throws Exception {
       if (queryMetadata instanceof PersistentQueryMetadata) {
         PersistentQueryMetadata persistentQueryMetadata = (PersistentQueryMetadata) queryMetadata;
         persistentQueryMetadata.getKafkaStreams().start();
-        ksqlEngine.getPersistentQueries()
-            .put(persistentQueryMetadata.getQueryId(), persistentQueryMetadata);
       } else {
         System.err.println("Ignoring statemenst: " + sql);
         System.err.println("Only CREATE statements can run in KSQL embedded mode.");

ksql-engine/src/main/java/io/confluent/ksql/KsqlEngine.java

@@ -26,6 +26,7 @@
 
 import java.io.Closeable;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -478,8 +479,14 @@ public boolean terminateQuery(final QueryId queryId, final boolean closeStreams)
     return true;
   }
 
-  public Map<QueryId, PersistentQueryMetadata> getPersistentQueries() {
-    return new HashMap<>(persistentQueries);
+  public PersistentQueryMetadata getPersistentQuery(QueryId queryId) {
+    return persistentQueries.get(queryId);
+  }
+
+  public Collection<PersistentQueryMetadata> getPersistentQueries() {
+    return Collections.unmodifiableList(
+        new ArrayList<>(
+            persistentQueries.values()));
   }
 
   public static List<String> getImmutableProperties() {

ksql-engine/src/test/java/io/confluent/ksql/KsqlContextTest.java

@@ -56,7 +56,6 @@ public void shouldRunSimpleStatements() throws Exception {
         (Collections.emptyList());
     expect(ksqlEngine.buildMultipleQueries(statement2, Collections.emptyMap()))
         .andReturn(getQueryMetadata(new QueryId("CSAS_BIGORDERS"), DataSource.DataSourceType.KSTREAM));
-    expect(ksqlEngine.getPersistentQueries()).andReturn(new HashMap<>());
     replay(ksqlEngine);
 
     KsqlContext ksqlContext = new KsqlContext(ksqlEngine);

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

@@ -17,16 +17,12 @@
 package io.confluent.ksql.rest.entity;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeName;
 import io.confluent.ksql.rest.server.computation.CommandId;
 
 import java.util.Map;
 import java.util.Objects;
 
-@JsonTypeName("currentStatus")
-@JsonSubTypes({})
 public class CommandStatusEntity extends KsqlEntity {
   private final CommandId commandId;
   private final CommandStatus commandStatus;

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

@@ -18,13 +18,9 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonSubTypes;
-import com.fasterxml.jackson.annotation.JsonTypeName;
 
 import java.util.Objects;
 
-@JsonTypeName("executionPlan")
-@JsonSubTypes({})
 public class ExecutionPlan extends KsqlEntity {
 
   private final String executionPlan;

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

@@ -20,8 +20,6 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonSubTypes;
-import com.fasterxml.jackson.annotation.JsonTypeName;
 
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.TopicPartition;
@@ -45,8 +43,6 @@
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.KsqlConstants;
 
-@JsonTypeName("kafka_topics")
-@JsonSubTypes({})
 public class KafkaTopicsList extends KsqlEntity {
 
   private final Collection<KafkaTopicInfo> topics;

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

@@ -27,9 +27,9 @@
     @JsonSubTypes.Type(value = CommandStatusEntity.class, name = "currentStatus"),
     @JsonSubTypes.Type(value = PropertiesList.class, name = "properties"),
     @JsonSubTypes.Type(value = Queries.class, name = "queries"),
-    @JsonSubTypes.Type(value = SourceDescriptionEntity.class, name = "description"),
-    @JsonSubTypes.Type(value = QueryDescriptionEntity.class, name = "query_description"),
-    @JsonSubTypes.Type(value = TopicDescription.class, name = "topic_description"),
+    @JsonSubTypes.Type(value = SourceDescriptionEntity.class, name = "sourceDescription"),
+    @JsonSubTypes.Type(value = QueryDescriptionEntity.class, name = "queryDescription"),
+    @JsonSubTypes.Type(value = TopicDescription.class, name = "topicDescription"),
     @JsonSubTypes.Type(value = StreamsList.class, name = "streams"),
     @JsonSubTypes.Type(value = TablesList.class, name = "tables"),
     @JsonSubTypes.Type(value = KsqlTopicsList.class, name = "ksql_topics"),