Dynamic listener: advertise default-listener-ip if dynamic-advertised-listener is not provided

Author: everesio

README.md

@@ -85,7 +85,7 @@ See:
           --debug-listen-address string                    Debug listen address (default "0.0.0.0:6060")
           --default-listener-ip string                     Default listener IP (default "127.0.0.1")
           --dial-address-mapping stringArray               Mapping of target broker address to new one (host:port,host:port). The mapping is performed during connection establishment
-          --dynamic-advertised-listener                    Advertised address for dynamic listeners (default "127.0.0.1")
+          --dynamic-advertised-listener                    Advertised address for dynamic listeners. If empty, default-listener-ip is used
           --dynamic-listeners-disable                      Disable dynamic listeners.
           --dynamic-sequential-min-port int                If set to non-zero, makes the dynamic listener use a sequential port starting with this value rather than a random port every time.
           --external-server-mapping stringArray            Mapping of Kafka server address to external address (host:port,host:port). A listener for the external address is not started

cmd/kafka-proxy/server.go

@@ -83,7 +83,7 @@ func init() {
 func initFlags() {
 	// proxy
 	Server.Flags().StringVar(&c.Proxy.DefaultListenerIP, "default-listener-ip", "127.0.0.1", "Default listener IP")
-	Server.Flags().StringVar(&c.Proxy.DynamicAdvertisedListener, "dynamic-advertised-listener", "127.0.0.1", "Advertised address for dynamic listeners")
+	Server.Flags().StringVar(&c.Proxy.DynamicAdvertisedListener, "dynamic-advertised-listener", "", "Advertised address for dynamic listeners. If empty, default-listener-ip is used")
 	Server.Flags().StringArrayVar(&bootstrapServersMapping, "bootstrap-server-mapping", []string{}, "Mapping of Kafka bootstrap server address to local address (host:port,host:port(,advhost:advport))")
 	Server.Flags().StringArrayVar(&externalServersMapping, "external-server-mapping", []string{}, "Mapping of Kafka server address to external address (host:port,host:port). A listener for the external address is not started")
 	Server.Flags().StringArrayVar(&dialAddressMapping, "dial-address-mapping", []string{}, "Mapping of target broker address to new one (host:port,host:port). The mapping is performed during connection establishment")

config/config.go

@@ -252,7 +252,6 @@ func NewConfig() *Config {
 	c.Http.HealthPath = "/health"
 
 	c.Proxy.DefaultListenerIP = "127.0.0.1"
-	c.Proxy.DynamicAdvertisedListener = "127.0.0.1"
 	c.Proxy.DisableDynamicListeners = false
 	c.Proxy.RequestBufferSize = 4096
 	c.Proxy.ResponseBufferSize = 4096

proxy/proxy.go

@@ -143,7 +143,7 @@ func (p *Listeners) ListenDynamicInstance(brokerAddress string) (string, int32,
 	defer p.lock.Unlock()
 	// double check
 	if v, ok := p.brokerToListenerConfig[brokerAddress]; ok {
-		return v.AdvertisedAddress, 0, nil
+		return util.SplitHostPort(v.AdvertisedAddress)
 	}
 
 	defaultListenerAddress := net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(p.dynamicSequentialMinPort))
@@ -158,9 +158,18 @@ func (p *Listeners) ListenDynamicInstance(brokerAddress string) (string, int32,
 	}
 	port := l.Addr().(*net.TCPAddr).Port
 	address := net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(port))
-	advertisedAddress := net.JoinHostPort(p.dynamicAdvertisedListener, fmt.Sprint(port))
+
+	dynamicAdvertisedListener := p.dynamicAdvertisedListener
+	if dynamicAdvertisedListener == "" {
+		dynamicAdvertisedListener = p.defaultListenerIP
+	}
+
+	advertisedAddress := net.JoinHostPort(dynamicAdvertisedListener, fmt.Sprint(port))
 	p.brokerToListenerConfig[brokerAddress] = config.ListenerConfig{BrokerAddress: brokerAddress, ListenerAddress: address, AdvertisedAddress: advertisedAddress}
-	return p.defaultListenerIP, int32(port), nil
+
+	logrus.Infof("Dynamic listener %s for broker %s advertised as %s", address, brokerAddress, advertisedAddress)
+
+	return dynamicAdvertisedListener, int32(port), nil
 }
 
 func (p *Listeners) ListenInstances(cfgs []config.ListenerConfig) (<-chan Conn, error) {