From 87f7c1a42191ed3461dfd5ae63200ed7801155d4 Mon Sep 17 00:00:00 2001
From: Stanislav Kozlovski <familyguyuser192@windowslive.com>
Date: Mon, 17 Jun 2019 15:38:33 +0100
Subject: [PATCH] MINOR: Fix system test logic to handle duplicates between 
 and

The changes are cherry-picked from commit 914ffa9dbef3c8ad6851b380276a1cb7c5aa4a0d which is present in 2.1+ versions
---
 tests/kafkatest/services/kafka/kafka.py | 36 ++++++++++++++++++-------
 1 file changed, 27 insertions(+), 9 deletions(-)

diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py
index 283bbe5df583a..b2289fd1fb19f 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -215,21 +215,39 @@ def set_protocol_and_port(self, node):
         self.advertised_listeners = ','.join(advertised_listeners)
 
     def prop_file(self, node):
-        cfg = KafkaConfig(**node.config)
-        cfg[config_property.ADVERTISED_HOSTNAME] = node.account.hostname
-        cfg[config_property.ZOOKEEPER_CONNECT] = self.zk_connect_setting()
+        self.set_protocol_and_port(node)
+
+        #load template configs as dictionary
+        config_template = self.render('kafka.properties', node=node, broker_id=self.idx(node),
+                                 security_config=self.security_config, num_nodes=self.num_nodes)
+
+        configs = dict( l.rstrip().split('=') for l in config_template.split('\n')
+                        if not l.startswith("#") and "=" in l )
+
+        #load specific test override configs
+        override_configs = KafkaConfig(**node.config)
+        override_configs[config_property.ADVERTISED_HOSTNAME] = node.account.hostname
+        override_configs[config_property.ZOOKEEPER_CONNECT] = self.zk_connect_setting()
 
         for prop in self.server_prop_overides:
-            cfg[prop[0]] = prop[1]
+            override_configs[prop[0]] = prop[1]
 
-        self.set_protocol_and_port(node)
+        #update template configs with test override configs
+        configs.update(override_configs)
 
-        # TODO - clean up duplicate configuration logic
-        prop_file = cfg.render()
-        prop_file += self.render('kafka.properties', node=node, broker_id=self.idx(node),
-                                 security_config=self.security_config, num_nodes=self.num_nodes)
+        prop_file = self.render_configs(configs)
         return prop_file
 
+    def render_configs(self, configs):
+        """Render self as a series of lines key=val\n, and do so in a consistent order. """
+        keys = [k for k in configs.keys()]
+        keys.sort()
+
+        s = ""
+        for k in keys:
+            s += "%s=%s\n" % (k, str(configs[k]))
+        return s
+
     def start_cmd(self, node):
         cmd = "export JMX_PORT=%d; " % self.jmx_port
         cmd += "export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % self.LOG4J_CONFIG