public static StatementExecutorResponse execute(
      final ConfiguredStatement<DropConnector> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    final String connectorName = statement.getStatement().getConnectorName();
    final boolean ifExists = statement.getStatement().getIfExists();
    final ConnectResponse<String> response =
        serviceContext.getConnectClient().delete(connectorName);

    if (response.error().isPresent()) {
      if (ifExists && response.httpCode() == HttpStatus.SC_NOT_FOUND) {
        return StatementExecutorResponse.handled(Optional.of(
            new WarningEntity(statement.getMaskedStatementText(),
                "Connector '" + connectorName + "' does not exist.")));
      } else {
        final String errorMsg = "Failed to drop connector: " + response.error().get();
        throw new KsqlRestException(EndpointResponse.create()
            .status(response.httpCode())
            .entity(new KsqlErrorMessage(Errors.toErrorCode(response.httpCode()), errorMsg))
            .build()
        );
      }
    }

    return StatementExecutorResponse.handled(Optional.of(
        new DropConnectorEntity(statement.getMaskedStatementText(), connectorName)));
  }
@Test
  public void shouldThrowOnError() {
    // Given:
    when(connectClient.delete(anyString()))
        .thenReturn(ConnectResponse.failure("Danger Mouse!", HttpStatus.SC_INTERNAL_SERVER_ERROR));

    // When:
    final KsqlRestException e = assertThrows(
        KsqlRestException.class,
        () -> DropConnectorExecutor.execute(
            DROP_CONNECTOR_CONFIGURED, mock(SessionProperties.class), null, serviceContext));
    final KsqlRestException eIfExists = assertThrows(
        KsqlRestException.class,
        () -> DropConnectorExecutor.execute(
            DROP_CONNECTOR_CONFIGURED, mock(SessionProperties.class), null, serviceContext));

    // Then:
    assertThat(e.getResponse().getStatus(), is(HttpStatus.SC_INTERNAL_SERVER_ERROR));
    final KsqlErrorMessage err = (KsqlErrorMessage) e.getResponse().getEntity();
    assertThat(err.getErrorCode(), is(Errors.toErrorCode(HttpStatus.SC_INTERNAL_SERVER_ERROR)));
    assertThat(err.getMessage(), containsString("Failed to drop connector: Danger Mouse!"));

    assertThat(eIfExists.getResponse().getStatus(), is(HttpStatus.SC_INTERNAL_SERVER_ERROR));
    final KsqlErrorMessage errIfExists = (KsqlErrorMessage) e.getResponse().getEntity();
    assertThat(errIfExists.getErrorCode(), is(Errors.toErrorCode(HttpStatus.SC_INTERNAL_SERVER_ERROR)));
    assertThat(errIfExists.getMessage(), containsString("Failed to drop connector: Danger Mouse!"));
  }