'Encode an OffsetFetchRequest struct. The request is encoded using
version 0 if from_kafka is false, indicating a request for Zookeeper
offsets. It is encoded using version 1 otherwise, indicating a request
for Kafka offsets.
Arguments:
group: string, the consumer group you are fetching offsets for
payloads: list of OffsetFetchRequestPayload
from_kafka: bool, default False, set True for Kafka-committed offsets'

@classmethod
def encode_offset_fetch_request(cls, group, payloads, from_kafka=False):

    version = (1 if from_kafka else 0)
    return kafka.protocol.commit.OffsetFetchRequest[version](consumer_group=group, topics=[(topic, list(topic_payloads.keys())) for (topic, topic_payloads) in six.iteritems(group_by_topic_and_partition(payloads))])
