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 shouldThrowIfConnectorTypeIsEmpty() {
    // Given:
    final CreateConnector createConnectorEmptyType = new CreateConnector(
        "foo", ImmutableMap.of("connector.class", new StringLiteral(" ")),
        Type.SOURCE, false);
    final ConfiguredStatement<CreateConnector> createConnectorEmptyTypeConfigured
        = ConfiguredStatement.of(PreparedStatement.of(
        "CREATE SOURCE CONNECTOR foo WITH ('connector.class'=' ');",
        createConnectorEmptyType), SessionConfig.of(CONFIG, ImmutableMap.of()));


    // When:
    final KsqlException e = assertThrows(
        KsqlException.class,
        () -> ConnectExecutor.validate(createConnectorEmptyTypeConfigured, mock(SessionProperties.class),
            null, serviceContext));

    // Then:
    assertThat(e.getMessage(), is("Validation error: Connector type cannot be empty"));
  }