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 shouldReturnWarningIfNotExist() {
    // Given:
    when(connectClient.delete(anyString()))
            .thenReturn(ConnectResponse.failure("Danger Mouse!", HttpStatus.SC_NOT_FOUND));

    // When:
    final Optional<KsqlEntity> entity = DropConnectorExecutor
            .execute(DROP_CONNECTOR_IF_EXISTS_CONFIGURED, mock(SessionProperties.class), null, serviceContext).getEntity();

    // Then:
    assertThat("Expected non-empty response", entity.isPresent());
    assertThat(entity.get(), instanceOf(WarningEntity.class));
  }