diff --git a/images/broker/Dockerfile b/images/broker/Dockerfile
index 3c3962a..b2c8b41 100644
--- a/images/broker/Dockerfile
+++ b/images/broker/Dockerfile
@@ -29,5 +29,7 @@ RUN wget http://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_${SCALA_VER
 # And we provide custom start up scripts
 ADD start_kafka /
 ADD start_zookeeper /
-RUN chmod +x start_kafka start_zookeeper \
-    && mkdir /logs/
+ADD create_topic /
+
+# Create structures required by our automation
+RUN mkdir /logs/
diff --git a/images/broker/create_topic b/images/broker/create_topic
new file mode 100755
index 0000000..ae99f01
--- /dev/null
+++ b/images/broker/create_topic
@@ -0,0 +1,14 @@
+#!/bin/bash
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic $1
diff --git a/images/broker/start_kafka b/images/broker/start_kafka
old mode 100644
new mode 100755
index 3a7c916..b69fa7f
--- a/images/broker/start_kafka
+++ b/images/broker/start_kafka
@@ -11,4 +11,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-/kafka/bin/kafka-server-start.sh /kafka/config/server.properties &
+/kafka/bin/kafka-server-start.sh /kafka.properties > /logs/kafka.log &
diff --git a/images/broker/start_zookeeper b/images/broker/start_zookeeper
old mode 100644
new mode 100755
index 39e67e6..778ec1a
--- a/images/broker/start_zookeeper
+++ b/images/broker/start_zookeeper
@@ -11,4 +11,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-/kafka/bin/zookeeper-server-start.sh /kafka/config/zookeeper.properties &
+/kafka/bin/zookeeper-server-start.sh /zookeeper.properties > /logs/zookeeper.log &
diff --git a/start.py b/start.py
index 4d4fd7a..1a061a5 100644
--- a/start.py
+++ b/start.py
@@ -11,6 +11,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import logging
 import tempfile
 import yaml
@@ -20,11 +21,41 @@
 from clusterdock.utils import wait_for_condition
 
 DEFAULT_NAMESPACE = 'clusterdock'
+ZOOKEEPER_PORT = 2181
+BROKER_PORT = 9092
 
 logger = logging.getLogger('clusterdock.{}'.format(__name__))
 
+def success(time):
+    logger.info('Conditions satisfied after %s seconds.', time)
+
+
+def failure(timeout):
+    raise TimeoutError('Timed out after {} seconds waiting.'.format(timeout))
+
+
+# Validate that Zookeeper is up and running by connecting using shell
+def validate_zookeeper(node, quiet):
+    return node.execute('/kafka/bin/zookeeper-shell.sh localhost:2181 ls /', quiet=quiet).exit_code == 0
+
+
+# Validate that Kafka is up by checking that all brokers are registered in zookeeper
+def validate_kafka(node, broker_count, quiet):
+    command = node.execute('/kafka/bin/zookeeper-shell.sh localhost:2181 ls /brokers/ids | tail -n 1', quiet=quiet)
+
+    if command.exit_code != 0:
+        return False
+
+    nodes = command.output
+    if not nodes.startswith('['):
+        return False
+
+    return len(json.loads(nodes)) == broker_count
+
 
 def main(args):
+    quiet = not args.verbose
+
     # Image name
     image = '{}/{}/topology_apache_kafka:kafka-{}-{}'.format(args.registry,
                                                              args.namespace or DEFAULT_NAMESPACE,
@@ -34,16 +65,60 @@ def main(args):
     # Nodes in the Kafka cluster
     nodes = [Node(hostname=hostname,
                   group='brokers',
-                  ports=[2181, 9092],
+                  ports=[ZOOKEEPER_PORT, BROKER_PORT],
                   image=image)
              for hostname in args.brokers]
 
     cluster = Cluster(*nodes)
     cluster.start(args.network, pull_images=args.always_pull)
 
-    # TODO: Add support for cluster mode (e.g. all nodes are part of the same cluster). Today we only
-    # start each node independently so they will all end up independent one-node clusters.
+    # Create distributed zookeeper configuration
+    zookeeper_config = ('tickTime=2000\n'
+                        'dataDir=/zookeeper\n'
+                        'clientPort=2181\n'
+                        'initLimit=5\n'
+                        'syncLimit=2\n')
+    for idx, node in enumerate(cluster):
+        zookeeper_config += 'server.{}={}:2888:3888\n'.format(idx, node.hostname)
 
+    # Start all zookeepers
+    for idx, node in enumerate(cluster):
+        logger.info('Starting Zookeeper on node {}'.format(node.hostname))
+        node.execute('mkdir -p /zookeeper')
+        node.put_file('/zookeeper/myid', str(idx))
+        node.put_file('/zookeeper.properties', zookeeper_config)
+        node.execute('/start_zookeeper &', detach=True)
+
+    # Validate that Zookeepr is alive from each node
     for node in cluster:
-        node.execute('/kafka/bin/zookeeper-server-start.sh /kafka/config/zookeeper.properties &', detach=True)
-        node.execute('/kafka/bin/kafka-server-start.sh /kafka/config/server.properties &', detach=True)
+        logger.info('Validating Zookeeper on node %s', node.hostname)
+        wait_for_condition(condition=validate_zookeeper,
+                           condition_args=[node, quiet],
+                           time_between_checks=3,
+                           timeout=60,
+                           success=success,
+                           failure=failure)
+
+    # Start all brokers
+    for idx, node in enumerate(cluster):
+        logger.info('Starting Kafka on node {}'.format(node.hostname))
+
+        kafka_config = node.get_file('/kafka/config/server.properties')
+        kafka_config = kafka_config.replace('broker.id=0', 'broker.id={}'.format(idx))
+        node.put_file('/kafka.properties', kafka_config)
+
+        node.execute('/start_kafka &', detach=True)
+
+    # Verify that all Kafka brokers up
+    logger.info('Waiting on all brokers to register in zookeeper')
+    wait_for_condition(condition=validate_kafka,
+                       condition_args=[nodes[0], len(nodes), quiet],
+                       time_between_checks=3,
+                       timeout=60,
+                       success=success,
+                       failure=failure)
+
+    # Automatically create topics
+    for topic in args.topics.split(','):
+        logger.info('Creating topic %s', topic)
+        nodes[0].execute('/create_topic {}'.format(topic), quiet=quiet)
diff --git a/topology.yaml b/topology.yaml
index a2e597a..c50ae66 100644
--- a/topology.yaml
+++ b/topology.yaml
@@ -35,3 +35,7 @@ start args:
         default: 2.11
         help: Scala version to use
         metavar: ver
+    --topics:
+        default: test
+        help: Comma-separated list of topics that should be auto-created
+        metavar: topic1,topic2