Provide a way to support Consumer#enforceRebalance

[sobychacko]

Currently, we can use the RecordInterceptor to access the consumer and trigger an enforceRebalance. Explore an option to see if we can support this operation as a first-class citizen from the container. Fore more details: https://kafka.apache.org/36/javadoc/org/apache/kafka/clients/consumer/Consumer.html#enforceRebalance()