Add topic and partition watching

[wvanbergen]

This adds watching partitions and topics for changes using Zookeeper watches.

Implementation

I also reworked the way we deal with Zookeeper watches. Zookeeper watches are inherently multithreaded, but I would like to hide this from the external API.

Thos most common use case is:

1. Get some information from Zookeeper and set a watch on it for changes.
2. Do something with the information returned from zookeeper.
3. Wait for the watch to be triggered until we start the next iteration.

The API I am using right now is:

loop do 
  # this is blocking until the zookeeper watch fires.
  topic.watch_partitions do |partitions|
    # this block gets called as soon as we know the partitions
  end
  # note that the zookeeper watch can fire before the block returns.
  # in that case we have to make sure not to deadlock

  # Next iteration of the loop
end

A small complication is that we sometimes want to set multiple zookeeper watches, and continue our processing when any of them fires. Right now, this is implemented by sharing a ConditionVariable instance:

cv = ConditionVariable.new

t1 = Thread.new do
  topic1.watch_partitions(cv: cv) do |partitions|
    # ...
  end
end

t2 = Thread.new do
  topic2.watch_partitions(cv: cv) do |partitions|
    # ...
  end
end

t1.join
t2.join

I am not a big fan of how this looks though. Any suggestions for a better API?