void handleStatement(final QueuedCommand queuedCommand) {
    throwIfNotConfigured();

    handleStatementWithTerminatedQueries(
        queuedCommand.getAndDeserializeCommand(commandDeserializer),
        queuedCommand.getAndDeserializeCommandId(),
        queuedCommand.getStatus(),
        Mode.EXECUTE,
        queuedCommand.getOffset(),
        false
    );
  }
@Test
  public void shouldThrowOnDropSourceWithoutPlan() {
    // Given:
    when(mockParser.parseSingleStatement("DROP STREAM"))
        .thenReturn(PreparedStatement.of("DROP STREAM", mock(DropStream.class)));

    final Command command = new Command(
        "DROP STREAM",
        emptyMap(),
        emptyMap(),
        Optional.empty());

    // When
    final Exception e = assertThrows(
        KsqlException.class,
        () -> handleStatement(statementExecutorWithMocks, command, COMMAND_ID, Optional.empty(), 0L)
    );

    // Then
    assertThat(e.getMessage(), containsString(
        "This version of ksqlDB does not support executing statements submitted prior to ksqlDB "
            + "0.8.0 or Confluent Platform ksqlDB 5.5. Please see the upgrading guide to upgrade."));
  }