diff --git a/docker/Dockerfile b/docker/Dockerfile
index 101b6bab..2a7c93bf 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -25,6 +25,7 @@ ENV \
     KAFKA_ALARM_HISTORY_BATCH_SIZE=1000 \
     KAFKA_ALARM_HISTORY_PROCESSORS=1 \
     KAFKA_ALARM_HISTORY_WAIT_TIME=15 \
+    KAFKA_LEGACY_CLIENT_ENABLED=true \
     KAFKA_METRICS_BATCH_SIZE=1000 \
     KAFKA_METRICS_PROCESSORS=1 \
     KAFKA_METRICS_WAIT_TIME=15 \
diff --git a/docker/README.rst b/docker/README.rst
index 286cdcd7..26c385b9 100644
--- a/docker/README.rst
+++ b/docker/README.rst
@@ -32,6 +32,7 @@ KAFKA_ALARM_HISTORY_BATCH_SIZE  1000              Kafka consumer takes messages
 KAFKA_ALARM_HISTORY_PROCESSORS  1                 Number of processes for alarm history topic
 KAFKA_ALARM_HISTORY_WAIT_TIME   15                Seconds to wait if the batch size is not reached
 KAFKA_EVENTS_ENABLE             false             Enable events persister
+KAFKA_LEGACY_CLIENT_ENABLED     true              Enable legacy Kafka client
 KAFKA_METRICS_BATCH_SIZE        1000              Kafka consumer takes messages in a batch
 KAFKA_METRICS_PROCESSORS        1                 Number of processes for metrics topic
 KAFKA_METRICS_WAIT_TIME         15                Seconds to wait if the batch size is not reached
diff --git a/docker/monasca-persister.conf.j2 b/docker/monasca-persister.conf.j2
index de3d5168..6c77b28c 100644
--- a/docker/monasca-persister.conf.j2
+++ b/docker/monasca-persister.conf.j2
@@ -47,6 +47,7 @@ max_buffer_size = 32768
 # Path in zookeeper for kafka consumer group partitioning algo
 zookeeper_path = /persister_partitions/alarm-state-transitions
 num_processors = {{ KAFKA_ALARM_HISTORY_PROCESSORS | default(1) }}
+legacy_kafka_client_enabled= {{ KAFKA_LEGACY_CLIENT_ENABLED | default(true) }}
 
 [kafka_events]
 # Comma separated list of Kafka broker host:port.
@@ -73,6 +74,7 @@ max_buffer_size = 32768
 # Path in zookeeper for kafka consumer group partitioning algo
 zookeeper_path = /persister_partitions/metrics
 num_processors = {{ KAFKA_METRICS_PROCESSORS | default(1) }}
+legacy_kafka_client_enabled= {{ KAFKA_LEGACY_CLIENT_ENABLED | default(true) }}
 
 {% if DATABASE_BACKEND | lower == 'cassandra' %}
 [cassandra]