Kafka errors on boot/shutdown sequence

[agis]

When performing a:

$ docker-compose up
# Ctrl+C
$ docker-compose down
$ docker-compose up

we get various warnings and errors in the logs:

kc2           | [2017-06-21 15:09:19,478] WARN Found a corrupted index file due to requirement failed: Corrupt index found, index file (/var/lib/kafka/__consumer_offsets-48/00000000000000000000.index) has non-zero size but the last offset
 is 0 which is no larger than the base offset 0.}. deleting /var/lib/kafka/__consumer_offsets-48/00000000000000000000.timeindex, /var/lib/kafka/__consumer_offsets-48/00000000000000000000.index and rebuilding index... (kafka.log.Log)

kc1           | java.lang.RuntimeException: A broker is already registered on the path /brokers/ids/1. This probably indicates that you either have configured a brokerid that is already in use, or else you have shutdown this broker and re
started it faster than the zookeeper timeout so it appears to be re-registering.
kc1           |         at kafka.utils.ZkUtils.registerBrokerInZk(ZkUtils.scala:408)
kc1           |         at kafka.utils.ZkUtils.registerBrokerInZk(ZkUtils.scala:394)
kc1           |         at kafka.server.KafkaHealthcheck.register(KafkaHealthcheck.scala:71)
kc1           |         at kafka.server.KafkaHealthcheck.startup(KafkaHealthcheck.scala:51)
kc1           |         at kafka.server.KafkaServer.startup(KafkaServer.scala:269)
kc1           |         at io.confluent.support.metrics.SupportedServerStartable.startup(SupportedServerStartable.java:102)
kc1           |         at io.confluent.support.metrics.SupportedKafka.main(SupportedKafka.java:49)


kc2           | [2017-06-21 15:09:20,827] FATAL [Kafka Server 2], Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kc2           | java.lang.RuntimeException: A broker is already registered on the path /brokers/ids/2. This probably indicates that you either have configured a brokerid that is already in use, or else you have shutdown this broker and re
started it faster than the zookeeper timeout so it appears to be re-registering.
kc2           |         at kafka.utils.ZkUtils.registerBrokerInZk(ZkUtils.scala:408)
kc2           |         at kafka.utils.ZkUtils.registerBrokerInZk(ZkUtils.scala:394)
kc2           |         at kafka.server.KafkaHealthcheck.register(KafkaHealthcheck.scala:71)
kc2           |         at kafka.server.KafkaHealthcheck.startup(KafkaHealthcheck.scala:51)
kc2           |         at kafka.server.KafkaServer.startup(KafkaServer.scala:269)
kc2           |         at io.confluent.support.metrics.SupportedServerStartable.startup(SupportedServerStartable.java:102)
kc2           |         at io.confluent.support.metrics.SupportedKafka.main(SupportedKafka.java:49)

Stopping kafkacluster_zoo2_1 ...
Stopping kc2 ...
Stopping kafkacluster_zoo1_1 ...
Stopping kafkacluster_zoo3_1 ...
Stopping kc1 ...
Exception in thread Thread-11:
Traceback (most recent call last):
  File "threading.py", line 810, in __bootstrap_inner
  File "threading.py", line 763, in run
  File "compose/cli/log_printer.py", line 190, in watch_events
  File "compose/project.py", line 343, in events
Stopping kafkacluster_zoo2_1 ... done
  File "json/__init__.py", line 338, in loads
Stopping kafkacluster_zoo1_1 ... done
Stopping kafkacluster_zoo3_1 ... done

These probably indicate that we're not performing the kafka/zookeeper shutdown sequence properly, leaving behind corrupted state.