added changes that brings in a new option with trivup command i.e a daemon flag(--d)

trivup/apps/KafkaBrokerApp.py

@@ -37,10 +37,10 @@
 import socket
 import time
 
-
 class KafkaBrokerApp (trivup.App):
     """ Kafka broker app
         Depends on ZookeeperApp (unless KRaft mode) """
+
     def __init__(self, cluster, conf=None, on=None):
         """
         @param cluster     Current cluster
@@ -73,6 +73,7 @@ def __init__(self, cluster, conf=None, on=None):
            * fdlimit - RLIMIT_NOFILE (or "max") (default: max)
            * conf - arbitary server.properties config as a list of strings.
            * realm - Kerberos realm to use when sasl_mechanisms contains GSSAPI
+           *
         """
         super(KafkaBrokerApp, self).__init__(cluster, conf=conf, on=on)
 
@@ -171,34 +172,61 @@ def __init__(self, cluster, conf=None, on=None):
 
         conf_blob.append(listener_map)
 
+
         def sort_listener(a):
             """ Sort listener_types list so that the PLAINTEXTs are first,
                 since the first listener is used by operational(). """
             if a.startswith('PLAINTEXT'):
                 return 0
             else:
                 return 1
-
-        # Allocate a port for each listener type
-        ports = [(x, trivup.TcpPortAllocator(self.cluster).next(
+
+        # If brokers ports are specified those would be used, if not it would fallback to TcpPortAllocator logic
+        if 'user_port' in self.conf and self.conf['user_port']:
+            ports = [(x, self.conf.get('user_port')) for x in sorted(set(listener_types), key=sort_listener)]
+            self.conf['port'] = self.conf.get('user_port') 
+            self.conf['address'] = '%s:%d' % (listener_host, self.conf.get('user_port'))
+        else :
+            ports = [(x, trivup.TcpPortAllocator(self.cluster).next(
             self, self.conf.get('port_base', self.conf.get('port', None))))
                  for x in sorted(set(listener_types), key=sort_listener)]
-        self.conf['port'] = ports[0][1]  # "Default" port
+            self.conf['port'] = ports[0][1]  # "Default" port
+            self.conf['address'] = '%s:%d' % (listener_host, self.conf['port'])
+
 
         if can_docker:
             # Add docker listener to allow services (e.g, SchemaRegistry) in
             # docker-containers to reach the on-host Kafka.
             docker_port = trivup.TcpPortAllocator(self.cluster).next(self)
             docker_host = '%s:%d' % (cluster.get_docker_host(), docker_port)
 
-        self.conf['address'] = '%s:%d' % (listener_host, self.conf['port'])
         # Create a listener for each port
         listeners = ['%s://%s:%d' % (x[0], "0.0.0.0", x[1]) for x in ports]
         if can_docker:
             listeners.append('%s://%s:%d' % ('DOCKER', "0.0.0.0", docker_port))
         self.conf['listeners'] = ','.join(listeners)
         if 'advertised_hostname' not in self.conf:
-            self.conf['advertised_hostname'] = self.conf['nodename']
+            # self.conf['advertised_hostname'] = self.conf['nodename']
+
+            def get_fqdn_from_resolvconf():
+                domain = None
+                with open("/etc/resolv.conf") as f:
+                    for line in f:
+                        if line.startswith("search") or line.startswith("domain"):
+                            parts = line.split()
+                            if len(parts) > 1:
+                                domain = parts[1]
+                                break
+
+                hostname = socket.gethostname()
+
+                if domain:
+                    return f"{hostname}.{domain}"
+                else:
+                    return hostname   
+
+            fqdn = get_fqdn_from_resolvconf()
+            self.conf['advertised_hostname'] = fqdn
         advertised_listeners = ['%s://%s:%d' %
                                 (x[0], self.conf['advertised_hostname'], x[1])
                                 for x in ports if x[0] != 'CONTROLLER']
@@ -462,4 +490,4 @@ def _add_simple_authorizer(self, conf_blob):
         if self.version[0] >= 3:
             conf_blob.append('authorizer.class.name=kafka.security.authorizer.AclAuthorizer')   # noqa: E501
         else:
-            conf_blob.append('authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer')   # noqa: E501
+            conf_blob.append('authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer')   # noqa: E501