class DistributedHerder extends AbstractHerder implements Runnable {

    @Override
    public void putConnectorConfig(final String connName, final Map<String, String> config, final boolean allowReplace,
                                   final Callback<Created<ConnectorInfo>> callback) {
        log.trace("Submitting connector config write request {}", connName);
        addRequest(
                new Callable<Void>() {
                    @Override
                    public Void call() throws Exception {
                        validateConnectorConfig(config, (error, configInfos) -> {
                            if (error != null) {
                                callback.onCompletion(error, null);
                                return;
                            }

                            // Complete the connector config write via another herder request in order to
                            // perform the write to the backing store (or forward to the leader) during
                            // the "external request" portion of the tick loop
                            addRequest(
                                    new Callable<Void>() {
                                        @Override
                                        public Void call() {
                                            if (maybeAddConfigErrors(configInfos, callback)) {
                                                return null;
                                            }

                                            log.trace("Handling connector config request {}", connName);
                                            if (!isLeader()) {
                                                callback.onCompletion(new NotLeaderException("Only the leader can set connector configs.", leaderUrl()), null);
                                                return null;
                                            }
                                            boolean exists = configState.contains(connName);
                                            if (!allowReplace && exists) {
                                                callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null);
                                                return null;
                                            }

                                            log.trace("Submitting connector config {} {} {}", connName, allowReplace, configState.connectors());
                                            configBackingStore.putConnectorConfig(connName, config);

                                            // Note that we use the updated connector config despite the fact that we don't have an updated
                                            // snapshot yet. The existing task info should still be accurate.
                                            ConnectorInfo info = new ConnectorInfo(connName, config, configState.tasks(connName),
                                                // validateConnectorConfig have checked the existence of CONNECTOR_CLASS_CONFIG
                                                connectorTypeForClass(config.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG)));
                                            callback.onCompletion(null, new Created<>(!exists, info));
                                            return null;
                                        }
                                    },
                                    forwardErrorCallback(callback)
                            );
                        });
                        return null;
                    }
                },
                forwardErrorCallback(callback)
        );
    }

    public  DistributedHerder(DistributedConfig config,
                             Time time,
                             Worker worker,
                             String kafkaClusterId,
                             StatusBackingStore statusBackingStore,
                             ConfigBackingStore configBackingStore,
                             String restUrl,
                             ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy);
      DistributedHerder(DistributedConfig config,
                      Worker worker,
                      String workerId,
                      String kafkaClusterId,
                      StatusBackingStore statusBackingStore,
                      ConfigBackingStore configBackingStore,
                      WorkerGroupMember member,
                      String restUrl,
                      ConnectMetrics metrics,
                      Time time,
                      ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy);

}