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

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());
   }
 }