diff --git a/ansible/action_plugins/merge_configs.py b/ansible/action_plugins/merge_configs.py
index 54463e6537..d20edf1da8 100644
--- a/ansible/action_plugins/merge_configs.py
+++ b/ansible/action_plugins/merge_configs.py
@@ -46,6 +46,12 @@ options:
     default: None
     required: True
     type: str
+  whitespace:
+    description:
+      - Whether whitespace characters should be used around equal signs
+    default: True
+    required: False
+    type: bool
 author: Sam Yaple
 '''
 
@@ -67,10 +73,11 @@ Merge multiple configs:
 
 class OverrideConfigParser(iniparser.BaseParser):
 
-    def __init__(self):
+    def __init__(self, whitespace=True):
         self._cur_sections = collections.OrderedDict()
         self._sections = collections.OrderedDict()
         self._cur_section = None
+        self._whitespace = ' ' if whitespace else ''
 
     def assignment(self, key, value):
         if self._cur_section is None:
@@ -107,12 +114,20 @@ class OverrideConfigParser(iniparser.BaseParser):
         def write_key_value(key, values):
             for v in values:
                 if not v:
-                    fp.write('{} =\n'.format(key))
+                    fp.write('{key}{ws}=\n'.format(
+                        key=key, ws=self._whitespace))
                 for index, value in enumerate(v):
                     if index == 0:
-                        fp.write('{} = {}\n'.format(key, value))
+                        fp.write('{key}{ws}={ws}{value}\n'.format(
+                            key=key,
+                            ws=self._whitespace,
+                            value=value))
                     else:
-                        fp.write('{}   {}\n'.format(len(key) * ' ', value))
+                        indent_size = len(key) + len(self._whitespace) * 2 + 1
+                        ws_indent = ' ' * indent_size
+                        fp.write('{ws_indent}{value}\n'.format(
+                            ws_indent=ws_indent,
+                            value=value))
 
         def write_section(section):
             for key, values in section.items():
@@ -158,7 +173,8 @@ class ActionModule(action.ActionBase):
         if not isinstance(sources, list):
             sources = [sources]
 
-        config = OverrideConfigParser()
+        config = OverrideConfigParser(
+            whitespace=self._task.args.get('whitespace', True))
 
         for source in sources:
             self.read_config(source, config)
@@ -179,6 +195,7 @@ class ActionModule(action.ActionBase):
 
             new_task = self._task.copy()
             new_task.args.pop('sources', None)
+            new_task.args.pop('whitespace', None)
 
             new_task.args.update(
                 dict(
diff --git a/ansible/roles/zookeeper/tasks/config.yml b/ansible/roles/zookeeper/tasks/config.yml
index abbf45ad0e..f289882432 100644
--- a/ansible/roles/zookeeper/tasks/config.yml
+++ b/ansible/roles/zookeeper/tasks/config.yml
@@ -31,6 +31,7 @@
       - "{{ role_path }}/templates/{{ item.key }}.cfg.j2"
       - "{{ node_custom_config }}/{{ item.key }}.cfg"
       - "{{ node_custom_config }}/{{ item.key }}/{{ inventory_hostname }}/{{ item.key }}.cfg"
+    whitespace: False
     dest: "{{ node_config_directory }}/{{ item.key }}/{{ item.key }}.cfg"
     mode: "0660"
   become: true
diff --git a/releasenotes/notes/zookeeper-cfg-syntax-02e93c01d6a24f35.yaml b/releasenotes/notes/zookeeper-cfg-syntax-02e93c01d6a24f35.yaml
new file mode 100644
index 0000000000..b24a4629a0
--- /dev/null
+++ b/releasenotes/notes/zookeeper-cfg-syntax-02e93c01d6a24f35.yaml
@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    Removes whitespace around equal signs in ``zookeeper.cfg`` which were
+    preventing the ``zkCleanup.sh`` script from running correctly.
diff --git a/tests/test_merge_config.py b/tests/test_merge_config.py
index ab5427c1ad..1a4cb3bc44 100644
--- a/tests/test_merge_config.py
+++ b/tests/test_merge_config.py
@@ -132,6 +132,29 @@ key2 = not_b
 
 '''
 
+# TESTC_NO_WHITESPACE is TESTA + TESTB without whitespace around equal signs
+TESTC_NO_WHITESPACE = '''[DEFAULT]
+key1=b
+     c
+key2=v3
+     v4
+     v5
+key3=v3
+key3=v4
+key4=v4
+key4=
+
+[b]
+b_key1=1
+b_key2=2
+
+[c]
+c_key1=
+c_key2=1 2 3
+       4 5 6
+
+'''
+
 
 class OverrideConfigParserTest(base.BaseTestCase):
 
@@ -178,3 +201,12 @@ class OverrideConfigParserTest(base.BaseTestCase):
         parser.write(output)
         self.assertEqual(TESTC_NO_DEFAULT_SECTION, output.getvalue())
         output.close()
+
+    def test_merge_no_whitespace(self):
+        parser = merge_configs.OverrideConfigParser(whitespace=False)
+        parser.parse(StringIO(TESTA))
+        parser.parse(StringIO(TESTB))
+        output = StringIO()
+        parser.write(output)
+        self.assertEqual(TESTC_NO_WHITESPACE, output.getvalue())
+        output.close()