prepare for rework

Author: everesio

cmd/kafka-proxy/server.go

@@ -89,7 +89,6 @@ func initFlags() {
 	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")
-	Server.Flags().BoolVar(&c.Proxy.DeterministicListeners, "deterministic-listeners", false, "Enable deterministic listeners (listener port = min port + broker id).")
 	Server.Flags().BoolVar(&c.Proxy.DisableDynamicListeners, "dynamic-listeners-disable", false, "Disable dynamic listeners.")
 	Server.Flags().IntVar(&c.Proxy.DynamicSequentialMinPort, "dynamic-sequential-min-port", 0, "If set to non-zero, makes the dynamic listener use a sequential port starting with this value rather than a random port every time.")
 

config/config.go

@@ -29,10 +29,6 @@ type ListenerConfig struct {
 	ListenerAddress   string
 	AdvertisedAddress string
 }
-type IdListenerConfig struct {
-	BrokerAddress string
-	Listener      net.Listener
-}
 type DialAddressMapping struct {
 	SourceAddress      string
 	DestinationAddress string
@@ -78,7 +74,6 @@ type Config struct {
 		DefaultListenerIP         string
 		BootstrapServers          []ListenerConfig
 		ExternalServers           []ListenerConfig
-		DeterministicListeners    bool
 		DialAddressMappings       []DialAddressMapping
 		DisableDynamicListeners   bool
 		DynamicAdvertisedListener string

proxy/protocol/responses.go

@@ -27,7 +27,7 @@ var (
 
 func createMetadataResponseSchemaVersions() []Schema {
 	metadataBrokerV0 := NewSchema("metadata_broker_v0",
-		&Mfield{Name: "node_id", Ty: TypeInt32},
+		&Mfield{Name: nodeKeyName, Ty: TypeInt32},
 		&Mfield{Name: hostKeyName, Ty: TypeStr},
 		&Mfield{Name: portKeyName, Ty: TypeInt32},
 	)
@@ -52,14 +52,14 @@ func createMetadataResponseSchemaVersions() []Schema {
 	)
 
 	metadataBrokerV1 := NewSchema("metadata_broker_v1",
-		&Mfield{Name: "node_id", Ty: TypeInt32},
+		&Mfield{Name: nodeKeyName, Ty: TypeInt32},
 		&Mfield{Name: hostKeyName, Ty: TypeStr},
 		&Mfield{Name: portKeyName, Ty: TypeInt32},
 		&Mfield{Name: "rack", Ty: TypeNullableStr},
 	)
 
 	metadataBrokerSchema9 := NewSchema("metadata_broker_schema9",
-		&Mfield{Name: "node_id", Ty: TypeInt32},
+		&Mfield{Name: nodeKeyName, Ty: TypeInt32},
 		&Mfield{Name: hostKeyName, Ty: TypeCompactStr},
 		&Mfield{Name: portKeyName, Ty: TypeInt32},
 		&Mfield{Name: "rack", Ty: TypeCompactNullableStr},
@@ -249,13 +249,13 @@ func createMetadataResponseSchemaVersions() []Schema {
 
 func createFindCoordinatorResponseSchemaVersions() []Schema {
 	findCoordinatorBrokerV0 := NewSchema("find_coordinator_broker_v0",
-		&Mfield{Name: "node_id", Ty: TypeInt32},
+		&Mfield{Name: nodeKeyName, Ty: TypeInt32},
 		&Mfield{Name: hostKeyName, Ty: TypeStr},
 		&Mfield{Name: portKeyName, Ty: TypeInt32},
 	)
 
 	findCoordinatorBrokerSchema9 := NewSchema("find_coordinator_broker_schema9",
-		&Mfield{Name: "node_id", Ty: TypeInt32},
+		&Mfield{Name: nodeKeyName, Ty: TypeInt32},
 		&Mfield{Name: hostKeyName, Ty: TypeCompactStr},
 		&Mfield{Name: portKeyName, Ty: TypeInt32},
 	)
@@ -341,7 +341,7 @@ func modifyMetadataResponse(decodedStruct *Struct, fn config.NetAddressMappingFu
 			}
 		}
 		if port != newPort {
-			err = broker.Replace(portKeyName, int32(newPort))
+			err = broker.Replace(portKeyName, newPort)
 			if err != nil {
 				return err
 			}

proxy/proxy.go

@@ -25,13 +25,11 @@ type Listeners struct {
 
 	listenFunc ListenFunc
 
-	deterministicListeners   bool
 	disableDynamicListeners  bool
 	dynamicSequentialMinPort int
 
-	brokerToListenerConfig     map[string]config.ListenerConfig
-	brokerIdToIdListenerConfig map[int32]config.IdListenerConfig
-	lock                       sync.RWMutex
+	brokerToListenerConfig map[string]config.ListenerConfig
+	lock                   sync.RWMutex
 }
 
 func NewListeners(cfg *config.Config) (*Listeners, error) {
@@ -66,19 +64,15 @@ func NewListeners(cfg *config.Config) (*Listeners, error) {
 		return nil, err
 	}
 
-	brokerIdToIdListenerConfig := make(map[int32]config.IdListenerConfig)
-
 	return &Listeners{
-		defaultListenerIP:          defaultListenerIP,
-		dynamicAdvertisedListener:  dynamicAdvertisedListener,
-		connSrc:                    make(chan Conn, 1),
-		brokerToListenerConfig:     brokerToListenerConfig,
-		brokerIdToIdListenerConfig: brokerIdToIdListenerConfig,
-		tcpConnOptions:             tcpConnOptions,
-		listenFunc:                 listenFunc,
-		deterministicListeners:     cfg.Proxy.DeterministicListeners,
-		disableDynamicListeners:    cfg.Proxy.DisableDynamicListeners,
-		dynamicSequentialMinPort:   cfg.Proxy.DynamicSequentialMinPort,
+		defaultListenerIP:         defaultListenerIP,
+		dynamicAdvertisedListener: dynamicAdvertisedListener,
+		connSrc:                   make(chan Conn, 1),
+		brokerToListenerConfig:    brokerToListenerConfig,
+		tcpConnOptions:            tcpConnOptions,
+		listenFunc:                listenFunc,
+		disableDynamicListeners:   cfg.Proxy.DisableDynamicListeners,
+		dynamicSequentialMinPort:  cfg.Proxy.DynamicSequentialMinPort,
 	}, nil
 }
 
@@ -132,49 +126,30 @@ func (p *Listeners) GetNetAddressMapping(brokerHost string, brokerPort int32, br
 
 	p.lock.RLock()
 	listenerConfig, ok := p.brokerToListenerConfig[brokerAddress]
-	idListenerConfig, brokerIdFound := p.brokerIdToIdListenerConfig[brokerId]
 	p.lock.RUnlock()
 
 	if ok {
-		logrus.Debugf("Address mappings broker=%s, listener=%s, advertised=%s", listenerConfig.BrokerAddress, listenerConfig.ListenerAddress, listenerConfig.AdvertisedAddress)
+		logrus.Debugf("Address mappings broker=%s, listener=%s, advertised=%s, node=%d", listenerConfig.BrokerAddress, listenerConfig.ListenerAddress, listenerConfig.AdvertisedAddress, brokerId)
 		return util.SplitHostPort(listenerConfig.AdvertisedAddress)
 	}
 	if !p.disableDynamicListeners {
-		if brokerIdFound {
-			logrus.Infof("Broker ID %d has a new advertised listener, closing existing dynamic listener", brokerId)
-			// Existing broker ID found, but with a different upstream broker
-			// Close existing listener, remove two mappings:
-			// * ID to removed upstream broker
-			// * removed upstream broker
-			idListenerConfig.Listener.Close()
-			p.lock.Lock()
-			delete(p.brokerIdToIdListenerConfig, brokerId)
-			delete(p.brokerToListenerConfig, idListenerConfig.BrokerAddress)
-			p.lock.Unlock()
-		}
 		logrus.Infof("Starting dynamic listener for broker %s", brokerAddress)
-		return p.ListenDynamicInstance(brokerAddress, brokerId)
+		return p.ListenDynamicInstance(brokerAddress)
 	}
 	return "", 0, fmt.Errorf("net address mapping for %s:%d was not found", brokerHost, brokerPort)
 }
 
-func (p *Listeners) ListenDynamicInstance(brokerAddress string, brokerId int32) (string, int32, error) {
+func (p *Listeners) ListenDynamicInstance(brokerAddress string) (string, int32, error) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 	// double check
 	if v, ok := p.brokerToListenerConfig[brokerAddress]; ok {
 		return util.SplitHostPort(v.AdvertisedAddress)
 	}
 
-	var defaultListenerAddress string
-
-	if p.deterministicListeners {
-		defaultListenerAddress = net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(p.dynamicSequentialMinPort+int(brokerId)))
-	} else {
-		defaultListenerAddress = net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(p.dynamicSequentialMinPort))
-		if p.dynamicSequentialMinPort != 0 {
-			p.dynamicSequentialMinPort += 1
-		}
+	defaultListenerAddress := net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(p.dynamicSequentialMinPort))
+	if p.dynamicSequentialMinPort != 0 {
+		p.dynamicSequentialMinPort += 1
 	}
 
 	cfg := config.ListenerConfig{ListenerAddress: defaultListenerAddress, BrokerAddress: brokerAddress}
@@ -192,7 +167,6 @@ func (p *Listeners) ListenDynamicInstance(brokerAddress string, brokerId int32)
 
 	advertisedAddress := net.JoinHostPort(dynamicAdvertisedListener, fmt.Sprint(port))
 	p.brokerToListenerConfig[brokerAddress] = config.ListenerConfig{BrokerAddress: brokerAddress, ListenerAddress: address, AdvertisedAddress: advertisedAddress}
-	p.brokerIdToIdListenerConfig[brokerId] = config.IdListenerConfig{BrokerAddress: brokerAddress, Listener: l}
 
 	logrus.Infof("Dynamic listener %s for broker %s advertised as %s", address, brokerAddress, advertisedAddress)