void handleLine(final String line) {
    final String trimmedLine = Optional.ofNullable(line).orElse("").trim();
    if (trimmedLine.isEmpty()) {
      return;
    }

    handleStatements(trimmedLine);
  }
@Test
  public void shouldThrowOnCCloudConnectorRequestWithoutApiKey() throws Exception {
    // Given:
    final KsqlRestClient mockRestClient = givenMockRestClient();
    when(mockRestClient.getIsCCloudServer()).thenReturn(true);
    when(mockRestClient.getHasCCloudApiKey()).thenReturn(false);

    // When:
    final Exception e = assertThrows(
        KsqlMissingCredentialsException.class,
        () -> localCli.handleLine("list connectors;")
    );

    // Then:
    assertThat(e.getMessage(), containsString("In order to use ksqlDB's connector "
        + "management capabilities with a Confluent Cloud ksqlDB server, launch the "
        + "ksqlDB command line with the additional flags '--confluent-api-key' and "
        + "'--confluent-api-secret' to pass a Confluent Cloud API key."));
    verify(mockRestClient, never()).makeConnectorRequest(anyString(), anyLong());
  }