Support for old ksql versions

1. Fixing ksql response column retrieval (existence check) (#3594)
2. checking "format" for streams commands if "valueFormat" not set

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java

@@ -52,7 +52,10 @@ public class KsqlApiClient {
     boolean error;
 
     public Optional<JsonNode> getColumnValue(List<JsonNode> row, String column) {
-      return Optional.ofNullable(row.get(columnNames.indexOf(column)));
+      int colIdx = columnNames.indexOf(column);
+      return colIdx >= 0
+          ? Optional.ofNullable(row.get(colIdx))
+          : Optional.empty();
     }
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlServiceV2.java

@@ -89,7 +89,14 @@ public class KsqlServiceV2 {
                       .name(resp.getColumnValue(row, "name").map(JsonNode::asText).orElse(null))
                       .topic(resp.getColumnValue(row, "topic").map(JsonNode::asText).orElse(null))
                       .keyFormat(resp.getColumnValue(row, "keyFormat").map(JsonNode::asText).orElse(null))
-                      .valueFormat(resp.getColumnValue(row, "valueFormat").map(JsonNode::asText).orElse(null)))
+                      .valueFormat(
+                          // for old versions (<0.13) "format" column is filled,
+                          // for new version "keyFormat" & "valueFormat" columns should be filled
+                          resp.getColumnValue(row, "valueFormat")
+                              .or(() -> resp.getColumnValue(row, "format"))
+                              .map(JsonNode::asText)
+                              .orElse(null))
+              )
               .collect(Collectors.toList()));
         });
   }