3.1 New Configuration Properties

Property	Type	Default	Description
`consumer.group.protocol`	string	`consumer`	Existing property. When set to `share`, the Connect worker creates a `KafkaShareConsumer` instead of a `KafkaConsumer` for sink tasks.

Property	Type	Default	Description
`consumer.override.group.protocol`	string	(inherited from worker)	Per-connector override. Set to `share` to opt a single connector into queue semantics.
`share.group.id`	string	`connect-<connector-name>`	The Share Group ID. Defaults to the same naming convention as consumer groups.
`share.acknowledgement.mode`	string	`explicit`	`explicit`: worker calls acknowledge(ACCEPT) after task.put() succeeds. `implicit`: acknowledgments are sent on the next poll() (simpler, lower latency, weaker guarantee).
`share.acquisition.lock.timeout.ms`	int	`30000`	Maximum time a record remains in ACQUIRED state before the broker releases it for re-delivery. Must be greater than the expected `task.put()` latency.
`share.delivery.semantics`	string	`at-least-once`	`at-least-once` or `exactly-once`. Exactly-once requires KIP-1289 and a transactional producer.
`share.max.delivery.count`	int	`5`	Maximum number of times a record can be re-delivered before being sent to the Dead Letter Queue (if configured). Maps to Share Group's `group.share.record.lock.partition.limit`.

...

The difference is entirely in the worker runtime:

Aspect	WorkerSinkTask (today)	WorkerShareSinkTask (proposed)
Consumer	`KafkaConsumer`	`KafkaShareConsumer`
Subscription	`consumer.subscribe(topics, rebalanceListener)`	`shareConsumer.subscribe(topics)`
Poll	`consumer.poll()`	`shareConsumer.poll()`
Offset tracking	`currentOffsets` map + `consumer.commitSync()`	Per-record `shareConsumer.acknowledge(record, ACCEPT)` + `shareConsumer.commitSync()`
Rebalance	`ConsumerRebalanceListener` calling `task.open()/close()`	No rebalances. `task.open()` called once at startup for all subscribed topics.
Failure handling	`RetriableException` -> pause consumer, retry batch	`RetriableException` -> `acknowledge(RELEASE)` for batch, records re-delivered by broker

4. Compatibility, Deprecation, and Migration Plan

...