Takes also now work. Multiple TODOs:

1. Java Serialization is broken. Use Avro to make sure things work.
2. Number of threads is currently too few. Increase the partition count in the config, make thread count configurable.
3. Clean up tests, reduce logging.