Merge pull request #30 from Yolean/kafka-011

Upgrade to Kafka 0.11 and test for open issues

25 files changed, 741 insertions, 481 deletions

diff --git a/10broker-config.yml b/10broker-config.yml
new file mode 100644
index 0000000..af0f037
--- /dev/null
+++ b/10broker-config.yml
@@ -0,0 +1,246 @@
+kind: ConfigMap
+metadata:
+  name: broker-config
+  namespace: kafka
+apiVersion: v1
+data:
+  init.sh: |-
+    #!/bin/bash
+    set -x
+
+    export KAFKA_BROKER_ID=${HOSTNAME##*-}
+    sed -i "s/\${KAFKA_BROKER_ID}/$KAFKA_BROKER_ID/" /etc/kafka/server.properties
+
+  server.properties: |-
+    # Licensed to the Apache Software Foundation (ASF) under one or more
+    # contributor license agreements.  See the NOTICE file distributed with
+    # this work for additional information regarding copyright ownership.
+    # The ASF licenses this file to You 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.
+
+    # see kafka.server.KafkaConfig for additional details and defaults
+
+    ############################# Server Basics #############################
+
+    # The id of the broker. This must be set to a unique integer for each broker.
+    broker.id=${KAFKA_BROKER_ID}
+
+    # Switch to enable topic deletion or not, default value is false
+    #delete.topic.enable=true
+
+    ############################# Socket Server Settings #############################
+
+    # The address the socket server listens on. It will get the value returned from
+    # java.net.InetAddress.getCanonicalHostName() if not configured.
+    #   FORMAT:
+    #     listeners = listener_name://host_name:port
+    #   EXAMPLE:
+    #     listeners = PLAINTEXT://your.host.name:9092
+    #listeners=PLAINTEXT://:9092
+
+    # Hostname and port the broker will advertise to producers and consumers. If not set,
+    # it uses the value for "listeners" if configured.  Otherwise, it will use the value
+    # returned from java.net.InetAddress.getCanonicalHostName().
+    #advertised.listeners=PLAINTEXT://your.host.name:9092
+
+    # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
+    #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
+
+    # The number of threads that the server uses for receiving requests from the network and sending responses to the network
+    num.network.threads=3
+
+    # The number of threads that the server uses for processing requests, which may include disk I/O
+    num.io.threads=8
+
+    # The send buffer (SO_SNDBUF) used by the socket server
+    socket.send.buffer.bytes=102400
+
+    # The receive buffer (SO_RCVBUF) used by the socket server
+    socket.receive.buffer.bytes=102400
+
+    # The maximum size of a request that the socket server will accept (protection against OOM)
+    socket.request.max.bytes=104857600
+
+
+    ############################# Log Basics #############################
+
+    # A comma seperated list of directories under which to store log files
+    log.dirs=/tmp/kafka-logs
+
+    # The default number of log partitions per topic. More partitions allow greater
+    # parallelism for consumption, but this will also result in more files across
+    # the brokers.
+    num.partitions=1
+
+    # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
+    # This value is recommended to be increased for installations with data dirs located in RAID array.
+    num.recovery.threads.per.data.dir=1
+
+    ############################# Internal Topic Settings  #############################
+    # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
+    # For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.
+    offsets.topic.replication.factor=1
+    transaction.state.log.replication.factor=1
+    transaction.state.log.min.isr=1
+
+    ############################# Log Flush Policy #############################
+
+    # Messages are immediately written to the filesystem but by default we only fsync() to sync
+    # the OS cache lazily. The following configurations control the flush of data to disk.
+    # There are a few important trade-offs here:
+    #    1. Durability: Unflushed data may be lost if you are not using replication.
+    #    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+    #    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
+    # The settings below allow one to configure the flush policy to flush data after a period of time or
+    # every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+    # The number of messages to accept before forcing a flush of data to disk
+    #log.flush.interval.messages=10000
+
+    # The maximum amount of time a message can sit in a log before we force a flush
+    #log.flush.interval.ms=1000
+
+    ############################# Log Retention Policy #############################
+
+    # The following configurations control the disposal of log segments. The policy can
+    # be set to delete segments after a period of time, or after a given size has accumulated.
+    # A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+    # from the end of the log.
+
+    # The minimum age of a log file to be eligible for deletion due to age
+    log.retention.hours=168
+
+    # A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
+    # segments don't drop below log.retention.bytes. Functions independently of log.retention.hours.
+    #log.retention.bytes=1073741824
+
+    # The maximum size of a log segment file. When this size is reached a new log segment will be created.
+    log.segment.bytes=1073741824
+
+    # The interval at which log segments are checked to see if they can be deleted according
+    # to the retention policies
+    log.retention.check.interval.ms=300000
+
+    ############################# Zookeeper #############################
+
+    # Zookeeper connection string (see zookeeper docs for details).
+    # This is a comma separated host:port pairs, each corresponding to a zk
+    # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+    # You can also append an optional chroot string to the urls to specify the
+    # root directory for all kafka znodes.
+    zookeeper.connect=localhost:2181
+
+    # Timeout in ms for connecting to zookeeper
+    zookeeper.connection.timeout.ms=6000
+
+
+    ############################# Group Coordinator Settings #############################
+
+    # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
+    # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
+    # The default value for this is 3 seconds.
+    # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
+    # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
+    group.initial.rebalance.delay.ms=0
+
+  log4j.properties: |-
+    # Licensed to the Apache Software Foundation (ASF) under one or more
+    # contributor license agreements.  See the NOTICE file distributed with
+    # this work for additional information regarding copyright ownership.
+    # The ASF licenses this file to You 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.
+
+    # Unspecified loggers and loggers with additivity=true output to server.log and stdout
+    # Note that INFO only applies to unspecified loggers, the log level of the child logger is used otherwise
+    log4j.rootLogger=INFO, stdout, kafkaAppender
+
+    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+    log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    log4j.appender.kafkaAppender=org.apache.log4j.DailyRollingFileAppender
+    log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH
+    log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log
+    log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
+    log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
+    log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH
+    log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
+    log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
+    log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    log4j.appender.requestAppender=org.apache.log4j.DailyRollingFileAppender
+    log4j.appender.requestAppender.DatePattern='.'yyyy-MM-dd-HH
+    log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log
+    log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
+    log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    log4j.appender.cleanerAppender=org.apache.log4j.DailyRollingFileAppender
+    log4j.appender.cleanerAppender.DatePattern='.'yyyy-MM-dd-HH
+    log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log
+    log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
+    log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    log4j.appender.controllerAppender=org.apache.log4j.DailyRollingFileAppender
+    log4j.appender.controllerAppender.DatePattern='.'yyyy-MM-dd-HH
+    log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log
+    log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
+    log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
+    log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
+    log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
+    log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
+    log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    # Change the two lines below to adjust ZK client logging
+    log4j.logger.org.I0Itec.zkclient.ZkClient=INFO
+    log4j.logger.org.apache.zookeeper=INFO
+
+    # Change the two lines below to adjust the general broker logging level (output to server.log and stdout)
+    log4j.logger.kafka=INFO
+    log4j.logger.org.apache.kafka=INFO
+
+    # Change to DEBUG or TRACE to enable request logging
+    log4j.logger.kafka.request.logger=WARN, requestAppender
+    log4j.additivity.kafka.request.logger=false
+
+    # Uncomment the lines below and change log4j.logger.kafka.network.RequestChannel$ to TRACE for additional output
+    # related to the handling of requests
+    #log4j.logger.kafka.network.Processor=TRACE, requestAppender
+    #log4j.logger.kafka.server.KafkaApis=TRACE, requestAppender
+    #log4j.additivity.kafka.server.KafkaApis=false
+    log4j.logger.kafka.network.RequestChannel$=WARN, requestAppender
+    log4j.additivity.kafka.network.RequestChannel$=false
+
+    log4j.logger.kafka.controller=TRACE, controllerAppender
+    log4j.additivity.kafka.controller=false
+
+    log4j.logger.kafka.log.LogCleaner=INFO, cleanerAppender
+    log4j.additivity.kafka.log.LogCleaner=false
+
+    log4j.logger.state.change.logger=TRACE, stateChangeAppender
+    log4j.additivity.state.change.logger=false
+
+    # Change to DEBUG to enable audit log for the authorizer
+    log4j.logger.kafka.authorizer.logger=WARN, authorizerAppender
+    log4j.additivity.kafka.authorizer.logger=false
diff --git a/10pvc.yml b/10pvc.yml
deleted file mode 100644
index 51de19c..0000000
--- a/10pvc.yml
+++ /dev/null
@@ -1,48 +0,0 @@
----
-kind: PersistentVolumeClaim
-apiVersion: v1
-metadata:
-  name: datadir-kafka-0
-  namespace: kafka
-spec:
-  accessModes:
-    - ReadWriteOnce
-  resources:
-    requests:
-      storage: 200Gi
-  selector:
-    matchLabels:
-      app: kafka
-      podindex: "0"
----
-kind: PersistentVolumeClaim
-apiVersion: v1
-metadata:
-  name: datadir-kafka-1
-  namespace: kafka
-spec:
-  accessModes:
-    - ReadWriteOnce
-  resources:
-    requests:
-      storage: 200Gi
-  selector:
-    matchLabels:
-      app: kafka
-      podindex: "1"
----
-kind: PersistentVolumeClaim
-apiVersion: v1
-metadata:
-  name: datadir-kafka-2
-  namespace: kafka
-spec:
-  accessModes:
-    - ReadWriteOnce
-  resources:
-    requests:
-      storage: 200Gi
-  selector:
-    matchLabels:
-      app: kafka
-      podindex: "2"
diff --git a/30service.yml b/30service.yml
deleted file mode 100644
index 5403da2..0000000
--- a/30service.yml
+++ /dev/null
@@ -1,11 +0,0 @@
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: kafka
-  namespace: kafka
-spec:
-  ports:
-  - port: 9092
-  selector:
-    app: kafka
diff --git a/50kafka.yml b/50kafka.yml
index 8a262df..4404a6b 100644
--- a/50kafka.yml
+++ b/50kafka.yml
@@ -10,23 +10,57 @@ spec:
     metadata:
       labels:
         app: kafka
+      annotations:
     spec:
-      terminationGracePeriodSeconds: 10
+      terminationGracePeriodSeconds: 30
+      initContainers:
+      - name: init-config
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        command: ['/bin/bash', '/etc/kafka/init.sh']
+        volumeMounts:
+        - name: config
+          mountPath: /etc/kafka
       containers:
       - name: broker
-        image: solsson/kafka-persistent:0.10.1@sha256:0719b4688b666490abf4b32a3cc5c5da7bb2d6276b47377b35de5429f783e9c2
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        env:
+        - name: KAFKA_LOG4J_OPTS
+          value: -Dlog4j.configuration=file:/etc/kafka/log4j.properties
         ports:
         - containerPort: 9092
         command:
-        - sh
-        - -c
-        - "./bin/kafka-server-start.sh config/server.properties --override broker.id=$(hostname | awk -F'-' '{print $2}')"
+        - ./bin/kafka-server-start.sh
+        - /etc/kafka/server.properties
+        - --override
+        -   zookeeper.connect=zookeeper:2181
+        - --override
+        -   log.retention.hours=-1
+        - --override
+        -   log.dirs=/var/lib/kafka/data/topics
+        - --override
+        -   auto.create.topics.enable=false
+        resources:
+          requests:
+            cpu: 100m
+            memory: 512Mi
+        livenessProbe:
+          exec:
+            command:
+            - /bin/sh
+            - -c
+            - 'echo "" | nc -w 1 127.0.0.1 9092'
         volumeMounts:
-        - name: datadir
-          mountPath: /opt/kafka/data
+        - name: config
+          mountPath: /etc/kafka
+        - name: data
+          mountPath: /var/lib/kafka/data
+      volumes:
+      - name: config
+        configMap:
+          name: broker-config
   volumeClaimTemplates:
   - metadata:
-      name: datadir
+      name: data
     spec:
       accessModes: [ "ReadWriteOnce" ]
       resources:
diff --git a/README.md b/README.md
index 492af46..9853d12 100644
--- a/README.md
+++ b/README.md
@@ -1,83 +1,61 @@
 
-# Kafka as Kubernetes StatefulSet
 
-Example of three Kafka brokers depending on five Zookeeper instances.
+# Kafka on Kubernetes
 
-To get consistent service DNS names `kafka-N.broker.kafka`(`.svc.cluster.local`), run everything in a [namespace](http://kubernetes.io/docs/admin/namespaces/walkthrough/):
-```
-kubectl create -f 00namespace.yml
-```
+Transparent Kafka setup that you can grow with.
+Good for both experiments and production.
 
-## Set up volume claims
+How to use:
+ * Run a Kubernetes cluster, [minikube](https://github.com/kubernetes/minikube) or real.
+ * Quickstart: use the `kubectl apply`s below.
+ * Kafka for real: fork and have a look at [addon](https://github.com/Yolean/kubernetes-kafka/labels/addon)s.
+ * Join the discussion in issues and PRs.
 
-You may add [storage class](http://kubernetes.io/docs/user-guide/persistent-volumes/#storageclasses)
-to the kafka StatefulSet declaration to enable automatic volume provisioning.
+Why?
+See for yourself, but we think this project gives you better adaptability than [helm](https://github.com/kubernetes/helm) [chart](https://github.com/kubernetes/charts/tree/master/incubator/kafka)s. No single readable readme or template can properly introduce both Kafka and Kubernets.
+Back when we read [Newman](http://samnewman.io/books/building_microservices/) we were beginners with both.
+Now we've read [Kleppmann](http://dataintensive.net/), [Confluent](https://www.confluent.io/blog/) and [SRE](https://landing.google.com/sre/book.html) and enjoy this "Streaming Platform" lock-in :smile:.
 
-Alternatively create [PV](http://kubernetes.io/docs/user-guide/persistent-volumes/#persistent-volumes)s and [PVC](http://kubernetes.io/docs/user-guide/persistent-volumes/#persistentvolumeclaims)s manually. For example in Minikube.
+## What you get
 
-```
-./bootstrap/pv.sh
-kubectl create -f ./bootstrap/pvc.yml
-# check that claims are bound
-kubectl get pvc
-```
+Keep an eye on `kubectl --namespace kafka get pods -w`.
+
+The goal is to provide [Bootstrap servers](http://kafka.apache.org/documentation/#producerconfigs): `kafka-0.broker.kafka.svc.cluster.local:9092,kafka-1.broker.kafka.svc.cluster.local:9092,kafka-2.broker.kafka.svc.cluster.local:9092`
+`
 
-## Set up Zookeeper
+Zookeeper at `zookeeper.kafka.svc.cluster.local:2181`.
 
-There is a Zookeeper+StatefulSet [blog post](http://blog.kubernetes.io/2016/12/statefulset-run-scale-stateful-applications-in-kubernetes.html) and [example](https://github.com/kubernetes/contrib/tree/master/statefulsets/zookeeper),
-but it appears tuned for workloads heavier than Kafka topic metadata.
+## Start Zookeeper
 
-The Kafka book (Definitive Guide, O'Reilly 2016) recommends that Kafka has its own Zookeeper cluster,
-so we use the [official docker image](https://hub.docker.com/_/zookeeper/)
-but with a [startup script change to guess node id from hostname](https://github.com/solsson/zookeeper-docker/commit/df9474f858ad548be8a365cb000a4dd2d2e3a217).
+The [Kafka book](https://www.confluent.io/resources/kafka-definitive-guide-preview-edition/) recommends that Kafka has its own Zookeeper cluster with at least 5 instances.
 
-Zookeeper runs as a [Deployment](http://kubernetes.io/docs/user-guide/deployments/) without persistent storage:
 ```
-kubectl create -f ./zookeeper/
+kubectl apply -f ./zookeeper/
 ```
 
-If you lose your zookeeper cluster, kafka will be unaware that persisted topics exist.
-The data is still there, but you need to re-create topics.
+To support automatic migration in the face of availability zone unavailability we mix persistent and ephemeral storage.
 
 ## Start Kafka
 
-Assuming you have your PVCs `Bound`, or enabled automatic provisioning (see above), go ahead and:
-
 ```
-kubectl create -f ./
+kubectl apply -f ./
 ```
 
 You might want to verify in logs that Kafka found its own DNS name(s) correctly. Look for records like:
 ```
-kubectl logs kafka-0 | grep "Registered broker"
+kubectl -n kafka logs kafka-0 | grep "Registered broker"
 # INFO Registered broker 0 at path /brokers/ids/0 with addresses: PLAINTEXT -> EndPoint(kafka-0.broker.kafka.svc.cluster.local,9092,PLAINTEXT)
 ```
 
-## Testing manually
-
-There's a Kafka pod that doesn't start the server, so you can invoke the various shell scripts.
-```
-kubectl create -f test/99testclient.yml
-```
-
-See `./test/test.sh` for some sample commands.
-
-## Automated test, while going chaosmonkey on the cluster
-
-This is WIP, but topic creation has been automated. Note that as a [Job](http://kubernetes.io/docs/user-guide/jobs/), it will restart if the command fails, including if the topic exists :(
-```
-kubectl create -f test/11topic-create-test1.yml
-```
-
-Pods that keep consuming messages (but they won't exit on cluster failures)
-```
-kubectl create -f test/21consumer-test1.yml
-```
+That's it. Just add business value :wink:.
+For clients we tend to use [librdkafka](https://github.com/edenhill/librdkafka)-based drivers like [node-rdkafka](https://github.com/Blizzard/node-rdkafka).
+To use [Kafka Connect](http://kafka.apache.org/documentation/#connect) and [Kafka Streams](http://kafka.apache.org/documentation/streams/) you may want to take a look at our [sample](https://github.com/solsson/dockerfiles/tree/master/connect-files) [Dockerfile](https://github.com/solsson/dockerfiles/tree/master/streams-logfilter)s.
+Don't forget the [addon](https://github.com/Yolean/kubernetes-kafka/labels/addon)s.
 
-## Teardown & cleanup
+# Tests
 
-Testing and retesting... delete the namespace. PVs are outside namespaces so delete them too.
 ```
-kubectl delete namespace kafka
-rm -R ./data/ && kubectl delete pv datadir-kafka-0 datadir-kafka-1 datadir-kafka-2
+kubectl apply -f test/
+# Anything that isn't READY here is a failed test
+kubectl get pods -l test-target=kafka,test-type=readiness -w --all-namespaces
 ```
diff --git a/bootstrap/pv-template.yml b/bootstrap/pv-template.yml
deleted file mode 100644
index e58bfb2..0000000
--- a/bootstrap/pv-template.yml
+++ /dev/null
@@ -1,45 +0,0 @@
----
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: datadir-kafka-0
-  labels:
-    app: kafka
-    podindex: "0"
-spec:
-  accessModes:
-  - ReadWriteOnce
-  capacity:
-    storage: 100Mi
-  hostPath:
-    path: /tmp/k8s-data/datadir-kafka-0
----
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: datadir-kafka-1
-  labels:
-    app: kafka
-    podindex: "1"
-spec:
-  accessModes:
-  - ReadWriteOnce
-  capacity:
-    storage: 100Mi
-  hostPath:
-    path: /tmp/k8s-data/datadir-kafka-1
----
-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: datadir-kafka-2
-  labels:
-    app: kafka
-    podindex: "2"
-spec:
-  accessModes:
-  - ReadWriteOnce
-  capacity:
-    storage: 100Mi
-  hostPath:
-    path: /tmp/k8s-data/datadir-kafka-2
diff --git a/bootstrap/pv.sh b/bootstrap/pv.sh
deleted file mode 100755
index 78bf7f5..0000000
--- a/bootstrap/pv.sh
+++ /dev/null
@@ -1,11 +0,0 @@
-#!/bin/bash
-
-echo "Note that in for example GKE a PetSet will have PersistentVolume(s) and PersistentVolumeClaim(s) created for it automatically"
-
-dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && cd .. && pwd )"
-path="$dir/data"
-echo "Please enter a path where to store data during local testing: ($path)"
-read newpath
-[ -n "$newpath" ] && path=$newpath
-
-cat bootstrap/pv-template.yml | sed "s|/tmp/k8s-data|$path|" | kubectl create -f -
diff --git a/docker-kafka-persistent/Dockerfile b/docker-kafka-persistent/Dockerfile
deleted file mode 100644
index 8120056..0000000
--- a/docker-kafka-persistent/Dockerfile
+++ /dev/null
@@ -1,19 +0,0 @@
-
-FROM openjdk:8u102-jre
-
-ENV kafka_version=0.10.1.1
-ENV scala_version=2.11.8
-ENV kafka_bin_version=2.11-$kafka_version
-
-RUN curl -SLs "http://www.scala-lang.org/files/archive/scala-$scala_version.deb" -o scala.deb \
-	&& dpkg -i scala.deb \
-	&& rm scala.deb \
-	&& curl -SLs "http://www.apache.org/dist/kafka/$kafka_version/kafka_$kafka_bin_version.tgz" | tar -xzf - -C /opt \
-	&& mv /opt/kafka_$kafka_bin_version /opt/kafka
-
-WORKDIR /opt/kafka
-ENTRYPOINT ["bin/kafka-server-start.sh"]
-
-ADD config/server.properties config/
-
-CMD ["config/server.properties"]
diff --git a/docker-kafka-persistent/config/server.properties b/docker-kafka-persistent/config/server.properties
deleted file mode 100644
index 649a261..0000000
--- a/docker-kafka-persistent/config/server.properties
+++ /dev/null
@@ -1,124 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-
-# see kafka.server.KafkaConfig for additional details and defaults
-
-############################# Server Basics #############################
-
-# The id of the broker. This must be set to a unique integer for each broker.
-broker.id=0
-
-# Use https://github.com/Yolean/kafka-topic-client instead
-auto.create.topics.enable=false
-
-# Switch to enable topic deletion or not, default value is false
-delete.topic.enable=false
-
-############################# Socket Server Settings #############################
-
-# The address the socket server listens on. It will get the value returned from 
-# java.net.InetAddress.getCanonicalHostName() if not configured.
-#   FORMAT:
-#     listeners = security_protocol://host_name:port
-#   EXAMPLE:
-#     listeners = PLAINTEXT://your.host.name:9092
-#listeners=PLAINTEXT://:9092
-
-# Hostname and port the broker will advertise to producers and consumers. If not set, 
-# it uses the value for "listeners" if configured.  Otherwise, it will use the value
-# returned from java.net.InetAddress.getCanonicalHostName().
-#advertised.listeners=PLAINTEXT://your.host.name:9092
-
-# The number of threads handling network requests
-num.network.threads=3
-
-# The number of threads doing disk I/O
-num.io.threads=8
-
-# The send buffer (SO_SNDBUF) used by the socket server
-socket.send.buffer.bytes=102400
-
-# The receive buffer (SO_RCVBUF) used by the socket server
-socket.receive.buffer.bytes=102400
-
-# The maximum size of a request that the socket server will accept (protection against OOM)
-socket.request.max.bytes=104857600
-
-
-############################# Log Basics #############################
-
-# A comma seperated list of directories under which to store log files
-log.dirs=/opt/kafka/data/topics
-
-# The default number of log partitions per topic. More partitions allow greater
-# parallelism for consumption, but this will also result in more files across
-# the brokers.
-num.partitions=1
-
-# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
-# This value is recommended to be increased for installations with data dirs located in RAID array.
-num.recovery.threads.per.data.dir=1
-
-############################# Log Flush Policy #############################
-
-# Messages are immediately written to the filesystem but by default we only fsync() to sync
-# the OS cache lazily. The following configurations control the flush of data to disk.
-# There are a few important trade-offs here:
-#    1. Durability: Unflushed data may be lost if you are not using replication.
-#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
-#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
-# The settings below allow one to configure the flush policy to flush data after a period of time or
-# every N messages (or both). This can be done globally and overridden on a per-topic basis.
-
-# The number of messages to accept before forcing a flush of data to disk
-#log.flush.interval.messages=10000
-
-# The maximum amount of time a message can sit in a log before we force a flush
-#log.flush.interval.ms=1000
-
-############################# Log Retention Policy #############################
-
-# The following configurations control the disposal of log segments. The policy can
-# be set to delete segments after a period of time, or after a given size has accumulated.
-# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
-# from the end of the log.
-
-# The minimum age of a log file to be eligible for deletion
-log.retention.hours=-1
-
-# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
-# segments don't drop below log.retention.bytes.
-#log.retention.bytes=1073741824
-
-# The maximum size of a log segment file. When this size is reached a new log segment will be created.
-log.segment.bytes=1073741824
-
-# The interval at which log segments are checked to see if they can be deleted according
-# to the retention policies
-log.retention.check.interval.ms=300000
-
-############################# Zookeeper #############################
-
-# Zookeeper connection string (see zookeeper docs for details).
-# This is a comma separated host:port pairs, each corresponding to a zk
-# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
-# You can also append an optional chroot string to the urls to specify the
-# root directory for all kafka znodes.
-zookeeper.connect=zookeeper:2181
-
-# Timeout in ms for connecting to zookeeper
-zookeeper.connection.timeout.ms=6000
-
-
diff --git a/prod-yolean.sh b/prod-yolean.sh
new file mode 100755
index 0000000..fb48139
--- /dev/null
+++ b/prod-yolean.sh
@@ -0,0 +1,16 @@
+#!/bin/bash
+# Combines addons into what we 'kubectl apply -f' to production
+set -ex
+
+git fetch
+git checkout origin/kafka-011
+git checkout -b prod-yolean-$(date +"%Y%m%dT%H%M%S")
+
+for BRANCH in \
+  addon-storage-classes \
+  addon-metrics \
+  addon-rest \
+  addon-kube-events-topic
+do
+  git merge --no-ff $BRANCH -m "prod-yolean merge $BRANCH"
+done
diff --git a/test/00namespace.yml b/test/00namespace.yml
new file mode 100644
index 0000000..fbb6e0e
--- /dev/null
+++ b/test/00namespace.yml
@@ -0,0 +1,5 @@
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: test-kafka
diff --git a/test/11topic-create-test1.yml b/test/11topic-create-test1.yml
deleted file mode 100644
index fdb805e..0000000
--- a/test/11topic-create-test1.yml
+++ /dev/null
@@ -1,25 +0,0 @@
-apiVersion: batch/v1
-kind: Job
-metadata:
-  name: topic-create-test1
-  namespace: kafka
-spec:
-  template:
-    metadata:
-      name: topic-create-test1
-    spec:
-      containers:
-      - name: kafka
-        image: solsson/kafka:0.10.0.1
-        command:
-        - ./bin/kafka-topics.sh
-        - --zookeeper
-        - zookeeper:2181
-        - --create
-        - --topic
-        - test1
-        - --partitions
-        - "1"
-        - --replication-factor
-        - "1"
-      restartPolicy: Never
diff --git a/test/12topic-create-test2.yml b/test/12topic-create-test2.yml
deleted file mode 100644
index 45d9881..0000000
--- a/test/12topic-create-test2.yml
+++ /dev/null
@@ -1,25 +0,0 @@
-apiVersion: batch/v1
-kind: Job
-metadata:
-  name: topic-create-test2
-  namespace: kafka
-spec:
-  template:
-    metadata:
-      name: topic-create-test2
-    spec:
-      containers:
-      - name: kafka
-        image: solsson/kafka:0.10.0.1
-        command:
-        - ./bin/kafka-topics.sh
-        - --zookeeper
-        - zookeeper:2181
-        - --create
-        - --topic
-        - test2
-        - --partitions
-        - "1"
-        - --replication-factor
-        - "3"
-      restartPolicy: Never
diff --git a/test/21consumer-test1.yml b/test/21consumer-test1.yml
deleted file mode 100644
index aff5944..0000000
--- a/test/21consumer-test1.yml
+++ /dev/null
@@ -1,24 +0,0 @@
-apiVersion: extensions/v1beta1
-kind: Deployment
-metadata:
-  name: consumer-test1
-  namespace: kafka
-spec:
-  replicas: 1
-  template:
-    metadata:
-      labels:
-        app: consumer
-        scope: test
-        topic: test1
-    spec:
-      containers:
-      - name: kafka
-        image: solsson/kafka:0.10.0.1
-        command:
-        - ./bin/kafka-console-consumer.sh
-        - --zookeeper
-        - zookeeper:2181
-        - --topic
-        - test1
-        - --from-beginning
diff --git a/test/99testclient.yml b/test/99testclient.yml
deleted file mode 100644
index 3ffa63a..0000000
--- a/test/99testclient.yml
+++ /dev/null
@@ -1,15 +0,0 @@
-# Kafka image without the service, so you can run ./bin/ stuff
-# kubectl exec -ti testclient -- /bin/bash
-apiVersion: v1
-kind: Pod
-metadata:
-  name: testclient
-  namespace: kafka
-spec:
-  containers:
-  - name: kafka
-    image: solsson/kafka-persistent:0.10.1@sha256:0719b4688b666490abf4b32a3cc5c5da7bb2d6276b47377b35de5429f783e9c2
-    command:
-      - sh
-      - -c
-      - "exec tail -f /dev/null"
diff --git a/test/basic-produce-consume.yml b/test/basic-produce-consume.yml
new file mode 100644
index 0000000..fdacea0
--- /dev/null
+++ b/test/basic-produce-consume.yml
@@ -0,0 +1,89 @@
+---
+kind: ConfigMap
+metadata:
+  name: basic-produce-consume
+  namespace: test-kafka
+apiVersion: v1
+data:
+
+  setup.sh: |-
+    touch /tmp/testlog
+
+    ./bin/kafka-topics.sh --zookeeper $ZOOKEEPER \
+      --create --if-not-exists --topic test-basic-produce-consume \
+      --partitions 1 --replication-factor 1
+
+    # Despite the deprecation warning --zookeeper nothing is consumed when using --bootstrap-server
+    ./bin/kafka-console-consumer.sh --zookeeper $ZOOKEEPER --topic test-basic-produce-consume > /tmp/testconsumed &
+
+    tail -f /tmp/testlog
+
+  continue.sh: |-
+    exit 0
+
+  run.sh: |-
+    exec >> /tmp/testlog
+    exec 2>&1
+
+    unique=$(date -Ins)
+
+    echo "Test $unique" | ./bin/kafka-console-producer.sh --broker-list $BOOTSTRAP --topic test-basic-produce-consume
+    echo ""
+    tail -n 1 /tmp/testconsumed | grep $unique
+
+    # How to make this test fail:
+    #apt-get update && apt-get install -y --no-install-recommends procps
+    #pkill java
+
+    exit 0
+
+---
+apiVersion: apps/v1beta1
+kind: Deployment
+metadata:
+  name: basic-produce-consume
+  namespace: test-kafka
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        test-target: kafka
+        test-type: readiness
+    spec:
+      containers:
+      - name: testcase
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        env:
+        - name: BOOTSTRAP
+          value: kafka-0.broker.kafka.svc.cluster.local:9092,kafka-1.broker.kafka.svc.cluster.local:9092,kafka-2.broker.kafka.svc.cluster.local:9092
+        - name: ZOOKEEPER
+          value: zookeeper.kafka.svc.cluster.local:2181
+        # Test set up
+        command:
+        - /bin/bash
+        - -e
+        - /test/setup.sh
+        # Test run, again and again
+        readinessProbe:
+          exec:
+            command:
+            - /bin/bash
+            - -e
+            - /test/run.sh
+          # JVM start is slow, can we keep producer started and restore the default preriod 10s?
+          periodSeconds: 30
+        # Test quit on nonzero exit
+        livenessProbe:
+          exec:
+            command:
+            - /bin/bash
+            - -e
+            - /test/continue.sh
+        volumeMounts:
+        - name: config
+          mountPath: /test
+      volumes:
+      - name: config
+        configMap:
+          name: basic-produce-consume
diff --git a/test/basic-with-kafkacat.yml b/test/basic-with-kafkacat.yml
new file mode 100644
index 0000000..a8974e8
--- /dev/null
+++ b/test/basic-with-kafkacat.yml
@@ -0,0 +1,103 @@
+---
+kind: ConfigMap
+metadata:
+  name: basic-with-kafkacat
+  namespace: test-kafka
+apiVersion: v1
+data:
+
+  setup.sh: |-
+    touch /tmp/testlog
+    tail -f /tmp/testlog
+
+  continue.sh: |-
+    exit 0
+
+  run.sh: |-
+    exec >> /tmp/testlog
+    exec 2>&1
+
+    unique=$(date -Ins)
+
+    echo "Test $unique" | kafkacat -P -b $BOOTSTRAP -t test-basic-with-kafkacat -v
+    kafkacat -C -b $BOOTSTRAP -t test-basic-with-kafkacat -o -1 -e | grep $unique
+
+    exit 0
+
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: basic-with-kafkacat
+  namespace: test-kafka
+spec:
+  template:
+    spec:
+      containers:
+      - name: topic-create
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        command:
+        - ./bin/kafka-topics.sh
+        - --zookeeper
+        -   zookeeper.kafka.svc.cluster.local:2181
+        - --create
+        - --if-not-exists
+        - --topic
+        -   test-basic-with-kafkacat
+        - --partitions
+        -   "1"
+        - --replication-factor
+        -   "1"
+      restartPolicy: Never
+---
+apiVersion: apps/v1beta1
+kind: Deployment
+metadata:
+  name: basic-with-kafkacat
+  namespace: test-kafka
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        test-target: kafka
+        test-type: readiness
+    spec:
+      containers:
+      - name: testcase
+        # common test images
+        #image: solsson/curl@sha256:8b0927b81d10043e70f3e05e33e36fb9b3b0cbfcbccdb9f04fd53f67a270b874
+        image: solsson/kafkacat@sha256:1266d140c52cb39bf314b6f22b6d7a01c4c9084781bc779fdfade51214a713a8
+        #image: solsson/kubectl-kafkacat@sha256:3715a7ede3f168f677ee6faf311ff6887aff31f660cfeecad5d87b4f18516321
+        env:
+        - name: BOOTSTRAP
+          #value: kafka-0.broker.kafka.svc.cluster.local:9092,kafka-1.broker.kafka.svc.cluster.local:9092,kafka-2.broker.kafka.svc.cluster.local:9092
+          value: kafka-0.broker.kafka.svc.cluster.local:9092
+        - name: ZOOKEEPER
+          value: zookeeper.kafka.svc.cluster.local:2181
+        # Test set up
+        command:
+        - /bin/bash
+        - -e
+        - /test/setup.sh
+        # Test run, again and again
+        readinessProbe:
+          exec:
+            command:
+            - /bin/bash
+            - -e
+            - /test/run.sh
+        # Test quit on nonzero exit
+        livenessProbe:
+          exec:
+            command:
+            - /bin/bash
+            - -e
+            - /test/continue.sh
+        volumeMounts:
+        - name: config
+          mountPath: /test
+      volumes:
+      - name: config
+        configMap:
+          name: basic-with-kafkacat
diff --git a/test/test.sh b/test/test.sh
deleted file mode 100644
index bfc4a8f..0000000
--- a/test/test.sh
+++ /dev/null
@@ -1,34 +0,0 @@
-
-# List topics
-kubectl exec testclient -- ./bin/kafka-topics.sh --zookeeper zookeeper:2181 --list
-
-# Create topic
-kubectl exec testclient -- ./bin/kafka-topics.sh --zookeeper zookeeper:2181 --topic test1 --create --partitions 1 --replication-factor 1
-
-# Set one of your terminals to listen to messages on the test topic
-kubectl exec -ti testclient -- ./bin/kafka-console-consumer.sh --zookeeper zookeeper:2181 --topic test1 --from-beginning
-
-# Go ahead and produce messages
-echo "Write a message followed by enter, exit using Ctrl+C"
-kubectl exec -ti testclient -- ./bin/kafka-console-producer.sh --broker-list kafka-0.broker.kafka.svc.cluster.local:9092 --topic test1
-
-# Bootstrap even if two nodes are down (shorter name requires same namespace)
-kubectl exec -ti testclient -- ./bin/kafka-console-producer.sh --broker-list kafka-0.broker:9092,kafka-1.broker:9092,kafka-2.broker:9092 --topic test1
-
-# The following commands run in the pod
-kubectl exec -ti testclient -- /bin/bash
-
-# Topic 2, replicated
-./bin/kafka-topics.sh --zookeeper zookeeper:2181 --describe --topic test2
-
-./bin/kafka-verifiable-consumer.sh \
-  --broker-list=kafka-0.broker.kafka.svc.cluster.local:9092,kafka-1.broker.kafka.svc.cluster.local:9092 \
-  --topic=test2 --group-id=A --verbose
-
-# If a topic isn't available this producer will tell you
-# WARN Error while fetching metadata with correlation id X : {topicname=LEADER_NOT_AVAILABLE}
-# ... but with current config Kafka will auto-create the topic
-./bin/kafka-verifiable-producer.sh \
-  --broker-list=kafka-0.broker.kafka.svc.cluster.local:9092,kafka-1.broker.kafka.svc.cluster.local:9092 \
-  --value-prefix=1 --topic=test2 \
-  --acks=1 --throughput=1 --max-messages=10
diff --git a/update-kafka-image.sh b/update-kafka-image.sh
new file mode 100755
index 0000000..fae9c6f
--- /dev/null
+++ b/update-kafka-image.sh
@@ -0,0 +1,10 @@
+#!/bin/bash
+
+IMAGE=$1
+[ -z "$IMAGE" ] && echo "First argument should be the image to set" && exit 1
+
+[[ $IMAGE != solsson/kafka:* ]] && echo "Should be the full image identifier" && exit 1
+
+for F in ./ test/ zookeeper/; do
+  sed -i "s|image: solsson/kafka:.*|image: $IMAGE|" $F*.yml
+done
diff --git a/zookeeper/10zookeeper-config.yml b/zookeeper/10zookeeper-config.yml
new file mode 100644
index 0000000..e796b4b
--- /dev/null
+++ b/zookeeper/10zookeeper-config.yml
@@ -0,0 +1,37 @@
+kind: ConfigMap
+metadata:
+  name: zookeeper-config
+  namespace: kafka
+apiVersion: v1
+data:
+  init.sh: |-
+    #!/bin/bash
+    set -x
+
+    [ -z "$ID_OFFSET" ] && ID_OFFSET=1
+    export ZOOKEEPER_SERVER_ID=$((${HOSTNAME##*-} + $ID_OFFSET))
+    echo "${ZOOKEEPER_SERVER_ID:-1}" | tee /var/lib/zookeeper/data/myid
+    sed -i "s/server\.$ZOOKEEPER_SERVER_ID\=[a-z0-9.-]*/server.$ZOOKEEPER_SERVER_ID=0.0.0.0/" /etc/kafka/zookeeper.properties
+
+  zookeeper.properties: |-
+    tickTime=2000
+    dataDir=/var/lib/zookeeper/data
+    dataLogDir=/var/lib/zookeeper/log
+    clientPort=2181
+    initLimit=5
+    syncLimit=2
+    server.1=pzoo-0.pzoo:2888:3888:participant
+    server.2=pzoo-1.pzoo:2888:3888:participant
+    server.3=pzoo-2.pzoo:2888:3888:participant
+    server.4=zoo-0.zoo:2888:3888:participant
+    server.5=zoo-1.zoo:2888:3888:participant
+
+  log4j.properties: |-
+    log4j.rootLogger=INFO, stdout
+    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+    log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+    # Suppress connection log messages, three lines per livenessProbe execution
+    log4j.logger.org.apache.zookeeper.server.NIOServerCnxnFactory=WARN
+    log4j.logger.org.apache.zookeeper.server.NIOServerCnxn=WARN
diff --git a/zookeeper/20pzoo-service.yml b/zookeeper/20pzoo-service.yml
new file mode 100644
index 0000000..00c33e1
--- /dev/null
+++ b/zookeeper/20pzoo-service.yml
@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: pzoo
+  namespace: kafka
+spec:
+  ports:
+  - port: 2888
+    name: peer
+  - port: 3888
+    name: leader-election
+  clusterIP: None
+  selector:
+    app: zookeeper
+    storage: persistent
diff --git a/zookeeper/20zoo-service.yml b/zookeeper/21zoo-service.yml
index d15dcc6..93fb321 100644
--- a/zookeeper/20zoo-service.yml
+++ b/zookeeper/21zoo-service.yml
@@ -12,3 +12,4 @@ spec:
   clusterIP: None
   selector:
     app: zookeeper
+    storage: ephemeral
diff --git a/zookeeper/50pzoo.yml b/zookeeper/50pzoo.yml
new file mode 100644
index 0000000..f9d5c58
--- /dev/null
+++ b/zookeeper/50pzoo.yml
@@ -0,0 +1,74 @@
+apiVersion: apps/v1beta1
+kind: StatefulSet
+metadata:
+  name: pzoo
+  namespace: kafka
+spec:
+  serviceName: "pzoo"
+  replicas: 3
+  template:
+    metadata:
+      labels:
+        app: zookeeper
+        storage: persistent
+      annotations:
+    spec:
+      terminationGracePeriodSeconds: 10
+      initContainers:
+      - name: init-config
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        command: ['/bin/bash', '/etc/kafka/init.sh']
+        volumeMounts:
+        - name: config
+          mountPath: /etc/kafka
+        - name: data
+          mountPath: /var/lib/zookeeper/data
+      containers:
+      - name: zookeeper
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        env:
+        - name: KAFKA_LOG4J_OPTS
+          value: -Dlog4j.configuration=file:/etc/kafka/log4j.properties
+        command:
+        - ./bin/zookeeper-server-start.sh
+        - /etc/kafka/zookeeper.properties
+        ports:
+        - containerPort: 2181
+          name: client
+        - containerPort: 2888
+          name: peer
+        - containerPort: 3888
+          name: leader-election
+        resources:
+          requests:
+            cpu: 10m
+            memory: 100Mi
+        livenessProbe:
+          exec:
+            command:
+            - /bin/sh
+            - -c
+            - '[ "imok" = "$(echo ruok | nc -w 1 127.0.0.1 2181)" ]'
+        readinessProbe:
+          exec:
+            command:
+            - /bin/sh
+            - -c
+            - '[ "imok" = "$(echo ruok | nc -w 1 127.0.0.1 2181)" ]'
+        volumeMounts:
+        - name: config
+          mountPath: /etc/kafka
+        - name: data
+          mountPath: /var/lib/zookeeper/data
+      volumes:
+      - name: config
+        configMap:
+          name: zookeeper-config
+  volumeClaimTemplates:
+  - metadata:
+      name: data
+    spec:
+      accessModes: [ "ReadWriteOnce" ]
+      resources:
+        requests:
+          storage: 10Gi
diff --git a/zookeeper/50zoo.yml b/zookeeper/50zoo.yml
deleted file mode 100644
index 5cb7c02..0000000
--- a/zookeeper/50zoo.yml
+++ /dev/null
@@ -1,38 +0,0 @@
-apiVersion: apps/v1beta1
-kind: StatefulSet
-metadata:
-  name: zoo
-  namespace: kafka
-spec:
-  serviceName: "zoo"
-  replicas: 5
-  template:
-    metadata:
-      labels:
-        app: zookeeper
-    spec:
-      terminationGracePeriodSeconds: 10
-      containers:
-        - name: zookeeper
-          image: solsson/zookeeper-statefulset:3.4.9@sha256:d32b44b32009a69b3450a5216f459e504f1041f587596895219fc04cf22f5546
-          env:
-            - name: ZOO_SERVERS
-              value: server.1=zoo-0.zoo:2888:3888:participant server.2=zoo-1.zoo:2888:3888:participant server.3=zoo-2.zoo:2888:3888:participant server.4=zoo-3.zoo:2888:3888:participant server.5=zoo-4.zoo:2888:3888:participant
-          ports:
-            - containerPort: 2181
-              name: client
-            - containerPort: 2888
-              name: peer
-            - containerPort: 3888
-              name: leader-election
-          volumeMounts:
-            - name: datadir
-              mountPath: /data
-            # There's defaults in this folder, such as logging config
-            #- name: conf
-            #  mountPath: /conf
-      volumes:
-        #- name: conf
-        #  emptyDir: {}
-        - name: datadir
-          emptyDir: {}
diff --git a/zookeeper/51zoo.yml b/zookeeper/51zoo.yml
new file mode 100644
index 0000000..778567d
--- /dev/null
+++ b/zookeeper/51zoo.yml
@@ -0,0 +1,71 @@
+apiVersion: apps/v1beta1
+kind: StatefulSet
+metadata:
+  name: zoo
+  namespace: kafka
+spec:
+  serviceName: "zoo"
+  replicas: 2
+  template:
+    metadata:
+      labels:
+        app: zookeeper
+        storage: ephemeral
+      annotations:
+    spec:
+      terminationGracePeriodSeconds: 10
+      initContainers:
+      - name: init-config
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        command: ['/bin/bash', '/etc/kafka/init.sh']
+        env:
+        - name: ID_OFFSET
+          value: "4"
+        volumeMounts:
+        - name: config
+          mountPath: /etc/kafka
+        - name: data
+          mountPath: /var/lib/zookeeper/data
+      containers:
+      - name: zookeeper
+        image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
+        env:
+        - name: KAFKA_LOG4J_OPTS
+          value: -Dlog4j.configuration=file:/etc/kafka/log4j.properties
+        command:
+        - ./bin/zookeeper-server-start.sh
+        - /etc/kafka/zookeeper.properties
+        ports:
+        - containerPort: 2181
+          name: client
+        - containerPort: 2888
+          name: peer
+        - containerPort: 3888
+          name: leader-election
+        resources:
+          requests:
+            cpu: 10m
+            memory: 100Mi
+        livenessProbe:
+          exec:
+            command:
+            - /bin/sh
+            - -c
+            - '[ "imok" = "$(echo ruok | nc -w 1 127.0.0.1 2181)" ]'
+        readinessProbe:
+          exec:
+            command:
+            - /bin/sh
+            - -c
+            - '[ "imok" = "$(echo ruok | nc -w 1 127.0.0.1 2181)" ]'
+        volumeMounts:
+        - name: config
+          mountPath: /etc/kafka
+        - name: data
+          mountPath: /var/lib/zookeeper/data
+      volumes:
+      - name: config
+        configMap:
+          name: zookeeper-config
+      - name: data
+        emptyDir: {}