...

Example: The default setting for `max.poll.records` is 500 for all Kafka consumers. We would be able to modify the behavior for a particular worker by adding a At present, if we want to modify this consumer client configuration, we could add a configuration to the worker properties that reads "consumer.max.poll.records=1000" if we wish to double the maximum number of records returned in a single call to poll(). However, this would impact every single sink connector on that worker. With connector-level configurability of all client configurations, we can target a specific connectors and their "max.poll.records" property by adding the "consumer.max.poll.records" configuration to the connector properties instead of the worker properties.

Other useful scenarios include being able to set "consumer.auto.offset.reset" to modify the initial offset assignment for specific consumers, and "producer.enable.idempotence" to provide specific source connectors with exactly once delivery semantics to ensure that producer retries do not write duplicates of retried messages.

Compatibility, Deprecation, and Migration Plan

...