能否从自定义Kafka Source Connector或Task访问Worker配置？

Absolutely, you can access Worker-level configurations like key converters, schema registry URLs, and ZooKeeper endpoints in your custom Kafka Source Connector and its Tasks—let me break down how to do this properly:

1. In the Connector Implementation #

While the ConnectorContext only provides methods for triggering reconfiguration, you don’t need it to access Worker configs. Here’s what to do:

• Leverage the start() method’s props: When the Worker initializes your Connector, it merges relevant global Worker configurations (like key.converter, value.converter, schema.registry.url) into the Map<String, String> props passed to the start() method—unless your Connector’s own config explicitly overrides these values.
• Use a custom ConnectorConfig class: For cleaner access, define a custom config class extending AbstractConfig, and include references to Worker config constants (from WorkerConfig or related classes like AbstractKafkaSchemaSerDeConfig). This lets you safely retrieve typed values.

Example code snippet:

public class MyCustomSourceConnector extends SourceConnector {
    private MyConnectorConfig config;

    @Override
    public void start(Map<String, String> props) {
        this.config = new MyConnectorConfig(props);
        // Fetch Worker-level configs
        String keyConverterClass = config.getString(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG);
        String schemaRegistryUrl = config.getString(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG);
        String zkConnect = config.getString(WorkerConfig.ZOOKEEPER_CONNECT_CONFIG);
    }

    // Define your custom config class
    public static class MyConnectorConfig extends AbstractConfig {
        public MyConnectorConfig(Map<String, String> props) {
            super(configDef(), props);
        }

        public static ConfigDef configDef() {
            return new ConfigDef()
                // Add your Connector-specific configs here
                .withClientSslSupport()
                .withClientSaslSupport();
        }
    }
}

2. In the Task Implementation #

Tasks can also access Worker configs through two main paths:

• Inherit configs from the Connector: The Map<String, String> props passed to the Task’s start() method includes all configs from the Connector (including the merged Worker configs) unless you restrict them.
• Explicitly pass configs via taskConfigs(): If you need to ensure specific Worker configs are available to Tasks, explicitly add them to the config maps returned by your Connector’s taskConfigs() method.

Example of passing configs to Tasks:

@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
    List<Map<String, String>> taskConfigs = new ArrayList<>();
    Map<String, String> baseProps = config.originals();

    for (int i = 0; i < maxTasks; i++) {
        Map<String, String> taskProps = new HashMap<>(baseProps);
        // Explicitly pass a Worker config if needed
        taskProps.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, 
                      config.getString(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG));
        taskConfigs.add(taskProps);
    }
    return taskConfigs;
}

Important Note: Config Override Policy #

By default, some Worker configs might not be exposed to Connectors. To allow full access to Worker-level configs, set the Worker’s connector.client.config.override.policy property to All in your Connect worker configuration file (e.g., connect-distributed.properties). Be cautious with this setting, as it grants Connectors access to sensitive configurations like credentials.

内容的提问来源于stack exchange，提问作者Xiang Zhang