void initialize() {
        try {
            if (!isSourceConnector() && !isSinkConnector()) {
                throw new ConnectException("Connector implementations must be a subclass of either SourceConnector or SinkConnector");
            }
            log.debug("{} Initializing connector {}", this, connName);
            if (isSinkConnector()) {
                SinkConnectorConfig.validate(config);
                connector.initialize(new WorkerSinkConnectorContext());
            } else {
                Objects.requireNonNull(offsetStore, "Offset store cannot be null for source connectors");
                Objects.requireNonNull(offsetStorageReader, "Offset reader cannot be null for source connectors");
                offsetStore.start();
                connector.initialize(new WorkerSourceConnectorContext(offsetStorageReader));
            }
        } catch (Throwable t) {
            log.error("{} Error initializing connector", this, t);
            onFailure(t);
        }
    }
Fail the connector.
@param cause the cause of the failure; if null, the connector will not be failed
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java