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");
  }
@SuppressWarnings("unchecked")
  @Test
  public void shouldPassInCorrectArgsToConnectClientOnExecute() {
    // Given:
    givenCreationSuccess();

    // When:
    ConnectExecutor
        .execute(CREATE_CONNECTOR_CONFIGURED, mock(SessionProperties.class), null, serviceContext);

    // Then:
    verify(connectClient).create(eq("foo"),
        (Map<String, String>) and(
            argThat(hasEntry("connector.class", "FileStreamSource")),
            argThat(hasEntry("name", "foo"))));
  }