Change max namespace count limit to per poller type (#3608)

common/rpc/interceptor/namespace_count_limit.go

@@ -51,7 +51,7 @@ type (
 		tokens  map[string]int
 
 		sync.Mutex
-		namespaceToCount map[namespace.Name]*int32
+		activeTokensCount map[string]*int32
 	}
 )
 
@@ -68,8 +68,7 @@ func NewNamespaceCountLimitInterceptor(
 		logger:            logger,
 		countFn:           countFn,
 		tokens:            tokens,
-
-		namespaceToCount: make(map[namespace.Name]*int32),
+		activeTokensCount: make(map[string]*int32),
 	}
 }
 
@@ -84,13 +83,15 @@ func (ni *NamespaceCountLimitInterceptor) Intercept(
 	token := ni.tokens[methodName]
 	if token != 0 {
 		nsName := GetNamespace(ni.namespaceRegistry, req)
-		counter := ni.counter(nsName)
+		counter := ni.counter(nsName, methodName)
 		count := atomic.AddInt32(counter, int32(token))
 		defer atomic.AddInt32(counter, -int32(token))
 
 		handler := GetMetricsHandlerFromContext(ctx, ni.logger)
 		handler.Gauge(metrics.ServicePendingRequests.GetMetricName()).Record(float64(count))
 
+		// frontend.namespaceCount is applied per poller type temporarily to prevent
+		// one poller type to take all token waiting in the long poll.
 		if int(count) > ni.countFn(nsName.String()) {
 			return nil, ErrNamespaceCountLimitServerBusy
 		}
@@ -101,15 +102,24 @@ func (ni *NamespaceCountLimitInterceptor) Intercept(
 
 func (ni *NamespaceCountLimitInterceptor) counter(
 	namespace namespace.Name,
+	methodName string,
 ) *int32 {
+	key := ni.getTokenKey(namespace, methodName)
+
 	ni.Lock()
 	defer ni.Unlock()
 
-	count, ok := ni.namespaceToCount[namespace]
+	counter, ok := ni.activeTokensCount[key]
 	if !ok {
-		counter := int32(0)
-		count = &counter
-		ni.namespaceToCount[namespace] = count
+		counter = new(int32)
+		ni.activeTokensCount[key] = counter
 	}
-	return count
+	return counter
+}
+
+func (ni *NamespaceCountLimitInterceptor) getTokenKey(
+	namespace namespace.Name,
+	methodName string,
+) string {
+	return namespace.String() + "/" + methodName
 }