I think here is how that can work -
Say we enable a "custom.beginning.offsets" option in the zk consumer. What that means is that this particular group id has chosen to work with custom beginning offsets.
The requirement is to make sure that only one consumer in the group specifies this option. And if more than one consumer in a group tries to do the same, we throw an exception. This can be implemented via zookeeper locks, where the first consumer id that specifies this option writes the custom offsets in the /consumers/groups/[group_id]/custom_offsets path. Following that any other consumer trying to do the same will get back an exception stating the consumer id that currently has the lock.
On the first successful rebalance after startup, the consumer ids can check for this path and reset the consumed offset for the partitions they own to that value.
This is just a cursory explanation of the idea and many details would have to be worked out. This is very tricky with the current implementation of the consumer rebalancing logic. I feel that this will be much easier to implement if we move to the co-ordinator approach for the consumer rebalancing. I've been meaning to write up a proposal for that. Will write up my ideas and add some ideas for this JIRA to it.