...

3.2.2`Worker.baseConsumerConfigs()` (modified)

A new internal class in `org.apache.kafka.connect.runtime` that extends `WorkerTask` and drives the `SinkTask` using a `KafkaShareConsumer` instead of a `KafkaConsumer`.

This is the core runtime change.

The existing method that builds consumer properties is modified to detect `group.protocol=share` and construct `KafkaShareConsumer` configs instead of `KafkaConsumer` configs:

```java
// In Worker.java
static Map<String, Object> baseConsumerConfigs(...) {
    Map<String, Object> consumerProps = new HashMap<>();
    
    String groupProtocol = // resolve from worker + connector config
    
    if ("share".equals(groupProtocol)) {
        consumerProps.put(ShareConsumerConfig.GROUP_ID_CONFIG, 
            connConfig.getString("share.group.id", SinkUtils.consumerGroupId(connName)));
        // Share consumer specific configs
        consumerProps.put(ShareConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServers());
        // No ENABLE_AUTO_COMMIT -- share groups don't have this concept
        // No AUTO_OFFSET_RESET -- share groups start from the share partition start offset
    } else {
        // existing consumer group config path (unchanged)
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, SinkUtils.consumerGroupId(connName));
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // ...
    }
    return consumerProps;```
// New class: parallel to WorkerSinkTask but backed by ShareConsumer
class WorkerShareSinkTask extends WorkerTask<ConsumerRecord<byte[], byte[]>, SinkRecord> {
    private final ShareConsumer<byte[], byte[]> shareConsumer;
    private final SinkTask task;
    // ...
}
```

4. Compatibility, Deprecation, and Migration Plan

...