public EndpointResponse streamQuery(
      final KsqlSecurityContext securityContext,
      final KsqlRequest request,
      final CompletableFuture<Void> connectionClosedFuture,
      final Optional<Boolean> isInternalRequest,
      final MetricsCallbackHolder metricsCallbackHolder,
      final Context context
  ) {
    throwIfNotConfigured();
    activenessRegistrar.updateLastRequestTime();

    final PreparedStatement<?> statement = parseStatement(request);

    CommandStoreUtil.httpWaitForCommandSequenceNumber(
        commandQueue, request, commandQueueCatchupTimeout);

    return handleStatement(securityContext, request, statement, connectionClosedFuture,
        isInternalRequest, metricsCallbackHolder, context);
  }
@Test
  public void shouldSuggestAlternativesIfPrintTopicDoesNotExist() {
    // Given:
    final PrintTopic cmd = mock(PrintTopic.class);
    when(cmd.getTopic()).thenReturn("TEST_TOPIC");
    print = PreparedStatement.of("print", cmd);
    when(mockStatementParser.<PrintTopic>parseSingleStatement(any()))
        .thenReturn(print);

    when(mockKafkaTopicClient.isTopicExists(any())).thenReturn(false);
    when(mockKafkaTopicClient.listTopicNames()).thenReturn(ImmutableSet.of(
        "aTopic",
        "test_topic",
        "Test_Topic"
    ));

    // When:
    final KsqlRestException e = assertThrows(
        KsqlRestException.class,
        () -> testResource.streamQuery(
            securityContext,
            new KsqlRequest(PRINT_TOPIC, Collections.emptyMap(), Collections.emptyMap(), null),
            new CompletableFuture<>(),
            Optional.empty(),
            new MetricsCallbackHolder(),
            context
        )
    );

    // Then:
    assertThat(e, exceptionStatusCode(is(BAD_REQUEST.code())));
    assertThat(e, exceptionErrorMessage(errorMessage(containsString(
        "Could not find topic 'TEST_TOPIC', "
            + "or the KSQL user does not have permissions to list the topic. "
            + "Topic names are case-sensitive."
            + System.lineSeparator()
            + "Did you mean:"
    ))));
    assertThat(e, exceptionErrorMessage(errorMessage(containsString("\tprint test_topic;"
    ))));
    assertThat(e, exceptionErrorMessage(errorMessage(containsString("\tprint Test_Topic;"
    ))));
  }