Support kafka consumer groups

[jfallows]

This requires creating a new subtype of kafka streams in kafka.idl.

KafkaGroupBeginEx
consumerId
session timeout

DATA
(leader can change assignments)
assignments (payload)

KafkaGroupDataEx
leader
members

FLUSH ⇒ heartbeat now

KafkaGroupClientFactory
configured group instance id (env)
configured client id (env)
handles rebalance internally

KafkaMergedBeginEx
consumerId
session timeout

subscribes only to topic partitions as indicated by assignments
OffsetFetch to bootstrap initial partition offsets
ack triggers OffsetCommit

KafkaOffsetCacheServerFactory
fan-in handles offset commit gaps
perhaps gaps can be handled via tag buffers (newer protocol versions)
or committed_metadata (older protocol versions okay)

KafkaFetchCacheServerFactory
anycast instead of broadcast for consumer group
handle redelivery if client fails, potentially out of order