fix: improve print topic error message when topic does not exist (#3464)

* fix: improve print topic error message when topic does not exist

The message incorrectly stated that KSQL treats unquoted topic names as uppercase. This is not the case.

Also, the suggested alternative was just the opposite case to the supplied, which is fairly arbitrary, e.g. if the user issues `print PageView` and the topic does not exist the error message will suggest `pAGEvIEW`, even though no such topic exists.

With this change the error message will include any other topics that exist in the cluster with the same name,  but different case.

ksql-engine/src/test/java/io/confluent/ksql/materialization/ks/KsStateStoreTest.java

@@ -65,7 +65,7 @@ public class KsStateStoreTest {
       .build();
 
   @Rule
-  public final Timeout timeout = Timeout.seconds(1);
+  public final Timeout timeout = Timeout.seconds(10);
 
   @Rule
   public final ExpectedException expectedException = ExpectedException.none();

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResource.java

@@ -40,9 +40,11 @@
 import io.confluent.ksql.util.TransientQueryMetadata;
 import io.confluent.ksql.version.metrics.ActivenessRegistrar;
 import java.time.Duration;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
@@ -232,24 +234,29 @@ private Response handlePrintTopic(
       final PreparedStatement<PrintTopic> statement
   ) {
     final PrintTopic printTopic = statement.getStatement();
-    final String topicName = printTopic.getTopic().toString();
+    final String topicName = printTopic.getTopic();
 
     if (!serviceContext.getTopicClient().isTopicExists(topicName)) {
-      String reverseSuggestion = "";
-      final String nameReversedCase = reverseCase(topicName);
-      if (serviceContext.getTopicClient().isTopicExists(nameReversedCase)) {
-        reverseSuggestion = "Did you mean '" + nameReversedCase + "'?" + System.lineSeparator();
-      }
+      final Collection<String> possibleAlternatives =
+          findPossibleTopicMatches(topicName, serviceContext);
+
+      final String reverseSuggestion = possibleAlternatives.isEmpty()
+          ? ""
+          : possibleAlternatives.stream()
+              .map(name -> "\tprint " + name + ";")
+              .collect(Collectors.joining(
+              System.lineSeparator(),
+              System.lineSeparator() + "Did you mean:" + System.lineSeparator(),
+              ""
+          ));
+
       throw new KsqlRestException(
-          Errors.badRequest(String.format(
-              "Could not find topic '%s', "
-                  + "or the KSQL user does not have permissions to list the topic."
-                  + System.lineSeparator()
+          Errors.badRequest(
+              "Could not find topic '" + topicName + "', "
+                  + "or the KSQL user does not have permissions to list the topic. "
+                  + "Topic names are case-sensitive."
                   + reverseSuggestion
-                  + "KSQL will treat unquoted topic names as uppercase."
-                  + System.lineSeparator()
-                  + "To print a case-sensitive topic use quotes, for example: print \'Topic\';",
-              topicName)));
+          ));
     }
 
     final Map<String, Object> propertiesWithOverrides =
@@ -267,17 +274,13 @@ private Response handlePrintTopic(
     return Response.ok().entity(topicStreamWriter).build();
   }
 
-  private String reverseCase(final String str) {
-    final char[] chars = str.toCharArray();
-    for (int i = 0; i < chars.length; i++) {
-      final char c = chars[i];
-      if (Character.isUpperCase(c)) {
-        chars[i] = Character.toLowerCase(c);
-      } else {
-        chars[i] = Character.toUpperCase(c);
-      }
-    }
-    return new String(chars);
+  private static Collection<String> findPossibleTopicMatches(
+      final String topicName,
+      final ServiceContext serviceContext
+  ) {
+    return serviceContext.getTopicClient().listTopicNames().stream()
+        .filter(name -> name.equalsIgnoreCase(topicName))
+        .collect(Collectors.toSet());
   }
 }