public static StatementExecutorResponse execute(
      final ConfiguredStatement<CreateConnector> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    final CreateConnector createConnector = statement.getStatement();
    final ConnectClient client = serviceContext.getConnectClient();

    final Optional<KsqlEntity> connectorsResponse = handleIfNotExists(
        statement, createConnector, client);
    if (connectorsResponse.isPresent()) {
      return StatementExecutorResponse.handled(connectorsResponse);
    }

    final ConnectResponse<ConnectorInfo> response = client.create(
        createConnector.getName(),
        buildConnectorConfig(createConnector));

    if (response.datum().isPresent()) {
      return StatementExecutorResponse.handled(Optional.of(
          new CreateConnectorEntity(
              statement.getMaskedStatementText(),
              response.datum().get()
          )
      ));
    }

    if (response.error().isPresent()) {
      final String errorMsg = "Failed to create connector: " + response.error().get();
      throw new KsqlRestException(EndpointResponse.create()
          .status(response.httpCode())
          .entity(new KsqlErrorMessage(Errors.toErrorCode(response.httpCode()), errorMsg))
          .build()
      );
    }

    throw new IllegalStateException("Either response.datum() or response.error() must be present");
  }
@Test
  public void shouldReturnWarningOnExecuteWhenIfNotExistsSetConnectorExists() {
    //Given:
    givenConnectorExists();

    //When
    final Optional<KsqlEntity> entity = ConnectExecutor
        .execute(CREATE_DUPLICATE_CONNECTOR_CONFIGURED,
            mock(SessionProperties.class), null, serviceContext).getEntity();
    //Then
    assertThat("Expected non-empty response", entity.isPresent());
    assertThat(entity.get(), instanceOf(WarningEntity.class));
  }