Change river to:
  - Have the river discover the set of brokers that exist from zk
  - Discover the leader broker for a given topic/partition
  - Don't store the brokerUrl in the zk offset path
  - On startup keep trying to discover/connect to kafka until there is a registered broker
  - On startup when no offsets are stored in zk, set the initial offset based on the 'startFromNewestOffset' setting