private void addHerder(SourceAndTarget sourceAndTarget) {
        log.info("creating herder for " + sourceAndTarget.toString());
        Map<String, String> workerProps = config.workerConfig(sourceAndTarget);
        List<String> restNamespace;
        try {
            String encodedSource = encodePath(sourceAndTarget.source());
            String encodedTarget = encodePath(sourceAndTarget.target());
            restNamespace = Arrays.asList(encodedSource, encodedTarget);
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException("Unable to create encoded URL paths for source and target using UTF-8", e);
        }
        String workerId = sourceAndTarget.toString();
        Plugins plugins = new Plugins(workerProps);
        plugins.compareAndSwapWithDelegatingLoader();
        DistributedConfig distributedConfig = new DistributedConfig(workerProps);
        String kafkaClusterId = distributedConfig.kafkaClusterId();
        String clientIdBase = ConnectUtils.clientIdBase(distributedConfig);
        // Create the admin client to be shared by all backing stores for this herder
        Map<String, Object> adminProps = new HashMap<>(distributedConfig.originals());
        adminProps.put(CLIENT_ID_CONFIG, clientIdBase + "shared-admin");
        ConnectUtils.addMetricsContextProperties(adminProps, distributedConfig, kafkaClusterId);
        SharedTopicAdmin sharedAdmin = new SharedTopicAdmin(adminProps);
        KafkaOffsetBackingStore offsetBackingStore = new KafkaOffsetBackingStore(sharedAdmin, () -> clientIdBase,
                plugins.newInternalConverter(true, JsonConverter.class.getName(),
                        Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false")));
        offsetBackingStore.configure(distributedConfig);
        ConnectorClientConfigOverridePolicy clientConfigOverridePolicy = new AllConnectorClientConfigOverridePolicy();
        clientConfigOverridePolicy.configure(config.originals());
        Worker worker = new Worker(workerId, time, plugins, distributedConfig, offsetBackingStore, clientConfigOverridePolicy);
        WorkerConfigTransformer configTransformer = worker.configTransformer();
        Converter internalValueConverter = worker.getInternalValueConverter();
        StatusBackingStore statusBackingStore = new KafkaStatusBackingStore(time, internalValueConverter, sharedAdmin, clientIdBase);
        statusBackingStore.configure(distributedConfig);
        ConfigBackingStore configBackingStore = new KafkaConfigBackingStore(
                internalValueConverter,
                distributedConfig,
                configTransformer,
                sharedAdmin,
                clientIdBase);
        // Pass the shared admin to the distributed herder as an additional AutoCloseable object that should be closed when the
        // herder is stopped. MirrorMaker has multiple herders, and having the herder own the close responsibility is much easier than
        // tracking the various shared admin objects in this class.
        Herder herder = new MirrorHerder(config, sourceAndTarget, distributedConfig, time, worker,
                kafkaClusterId, statusBackingStore, configBackingStore,
                advertisedUrl, restClient, clientConfigOverridePolicy,
                restNamespace, sharedAdmin);
        herders.put(sourceAndTarget, herder);
    }
@param config    MM2 configuration from mm2.properties file
@param clusters  target clusters for this node. These must match cluster
                 aliases as defined in the config. If null or empty list,
                 uses all clusters in the config.
@param time      time source
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMaker.java