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

    if (checkForExistingConnector(statement, createConnector, client)) {
      final String errorMsg = String.format(
          "Connector %s already exists", createConnector.getName());
      throw new KsqlRestException(EndpointResponse.create()
          .status(HttpStatus.SC_CONFLICT)
          .entity(new KsqlErrorMessage(Errors.toErrorCode(HttpStatus.SC_CONFLICT), errorMsg))
          .build()
      );
    }

    final List<String> errors = validateConfigs(createConnector, client);
    if (!errors.isEmpty()) {
      final String errorMessage = "Validation error: " + String.join("\n", errors);
      throw new KsqlException(errorMessage);
    }

    return StatementExecutorResponse.handled(Optional.of(new CreateConnectorEntity(
        statement.getMaskedStatementText(),
        DUMMY_CREATE_RESPONSE
    )));
  }
@Test
  public void shouldThrowOnValidationError() {
    // Given:
    givenValidationError();

    // When / Then:
    assertThrows(
        KsqlException.class,
        () -> ConnectExecutor.validate(
            CREATE_CONNECTOR_CONFIGURED, mock(SessionProperties.class), null, serviceContext));
  }