static Map<String, Object> adminConfigs(String connName,
                                            String defaultClientId,
                                            WorkerConfig config,
                                            ConnectorConfig connConfig,
                                            Class<? extends Connector> connectorClass,
                                            ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
                                            String clusterId,
                                            ConnectorType connectorType) {
        Map<String, Object> adminProps = new HashMap<>();
        // Use the top-level worker configs to retain backwards compatibility with older releases which
        // did not require a prefix for connector admin client configs in the worker configuration file
        // Ignore configs that begin with "admin." since those will be added next (with the prefix stripped)
        // and those that begin with "producer." and "consumer.", since we know they aren't intended for
        // the admin client
        Map<String, Object> nonPrefixedWorkerConfigs = config.originals().entrySet().stream()
                .filter(e -> !e.getKey().startsWith("admin.")
                        && !e.getKey().startsWith("producer.")
                        && !e.getKey().startsWith("consumer."))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServers());
        adminProps.put(AdminClientConfig.CLIENT_ID_CONFIG, defaultClientId);
        adminProps.putAll(nonPrefixedWorkerConfigs);

        // Admin client-specific overrides in the worker config
        adminProps.putAll(config.originalsWithPrefix("admin."));

        // Connector-specified overrides
        Map<String, Object> adminOverrides =
                connectorClientConfigOverrides(connName, connConfig, connectorClass, ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX,
                        connectorType, ConnectorClientConfigRequest.ClientType.ADMIN,
                        connectorClientConfigOverridePolicy);
        adminProps.putAll(adminOverrides);

        //add client metrics.context properties
        ConnectUtils.addMetricsContextProperties(adminProps, config, clusterId);

        return adminProps;
    }
Using the admin principal for this connector, perform a round of zombie fencing that disables transactional producers
for the specified number of source tasks from sending any more records.
@param connName the name of the connector
@param numTasks the number of tasks to fence out
@param connProps the configuration of the connector; may not be null
@return a {@link KafkaFuture} that will complete when the producers have all been fenced out, or the attempt has failed
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java