fix(ReplicationBudgetManager): enable ReplicationBudgetManager by default (#7448)

<!-- Describe what has changed in this PR -->
**What changed?**
Enable ReplicationBudgetManager by default, removing the dynamic config
to enable it.

<!-- Tell your future self why have you made these changes -->
**Why?**
It can be disabled directly in the manager and it's disabled by default.
It also creates confusion and allows the replication cache to be enable
by shard only if the configs are not set properly. This could cause OOM
issues.

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
Ran simulation and checked logs locally

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**
This enables the budget manager by default but the budget manager
capacity is set to zero. It should be unless the capacity has been
explicitly set to not 0.

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

---------

Signed-off-by: fimanishi <[email protected]>

Author: fimanishi

common/cache/budget.go

@@ -318,6 +318,11 @@ func (m *manager) metricsLoop() {
 
 // updateMetrics emits current state metrics
 func (m *manager) updateMetrics() {
+	// Skip metrics if both capacity limits are disabled (set to 0)
+	if m.maxBytes() == 0 && m.maxCount() == 0 {
+		return
+	}
+
 	// Emit capacity metrics
 	capacityBytes := m.CapacityBytes()
 	// Only emit if not unlimited
@@ -344,6 +349,11 @@ func (m *manager) updateMetrics() {
 
 // emitHardCapExceeded logs when hard capacity limit is exceeded and increments counter
 func (m *manager) emitHardCapExceeded(cacheID, budgetType string, requested, available uint64) {
+	// Skip metrics if both capacity limits are disabled (set to 0)
+	if m.maxBytes() == 0 && m.maxCount() == 0 {
+		return
+	}
+
 	if m.scope != nil {
 		m.scope.IncCounter(metrics.BudgetManagerHardCapExceeded)
 	}
@@ -361,6 +371,11 @@ func (m *manager) emitHardCapExceeded(cacheID, budgetType string, requested, ava
 
 // emitSoftCapExceeded logs when soft cap is exceeded and increments counter
 func (m *manager) emitSoftCapExceeded(cacheID, budgetType string, requested, available uint64) {
+	// Skip metrics if both capacity limits are disabled (set to 0)
+	if m.maxBytes() == 0 && m.maxCount() == 0 {
+		return
+	}
+
 	if m.scope != nil {
 		m.scope.IncCounter(metrics.BudgetManagerSoftCapExceeded)
 	}
@@ -378,6 +393,11 @@ func (m *manager) emitSoftCapExceeded(cacheID, budgetType string, requested, ava
 
 // emitCapacityExceeded emits appropriate metrics based on the error type
 func (m *manager) emitCapacityExceeded(cacheID string, err error, requestedBytes uint64, requestedCount int64, capResult CapEnforcementResult) {
+	// Skip metrics if both capacity limits are disabled (set to 0)
+	if m.maxBytes() == 0 && m.maxCount() == 0 {
+		return
+	}
+
 	switch err {
 	case ErrBytesBudgetExceeded:
 		m.emitHardCapExceeded(cacheID, budgetTypeBytes, requestedBytes, capResult.AvailableBytes)

common/cache/budget_test.go

@@ -11,9 +11,11 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 
 	"github.com/uber/cadence/common/dynamicconfig/dynamicproperties"
 	"github.com/uber/cadence/common/log/testlogger"
+	metricsmocks "github.com/uber/cadence/common/metrics/mocks"
 )
 
 func TestBudgetManager_ReserveForCache(t *testing.T) {
@@ -2230,3 +2232,40 @@ func BenchmarkBudgetManager_HighContention(b *testing.B) {
 		}
 	})
 }
+
+func TestBudgetManager_DisabledManager_NoMetricsEmitted(t *testing.T) {
+	// Create a mock scope that will track all metric calls
+	mockScope := &metricsmocks.Scope{}
+	// Mock the Tagged call which returns itself
+	mockScope.On("Tagged", mock.Anything).Return(mockScope)
+
+	// Create manager with BOTH capacities disabled (set to 0)
+	mgr := NewBudgetManager(
+		"test-disabled-manager",
+		dynamicproperties.GetIntPropertyFn(0), // maxBytes = 0 (disabled)
+		dynamicproperties.GetIntPropertyFn(0), // maxCount = 0 (disabled)
+		AdmissionOptimistic,
+		0,
+		mockScope,
+		testlogger.New(t),
+		dynamicproperties.GetFloatPropertyFn(0.8),
+	)
+	defer mgr.Stop()
+
+	// Give time for any initial metrics to be emitted
+	time.Sleep(100 * time.Millisecond)
+
+	// Try to reserve capacity (should fail but shouldn't emit metrics)
+	err := mgr.(*manager).ReserveForCache("cache1", 100, 1)
+	assert.Error(t, err)
+
+	// Try to release capacity (should not emit metrics)
+	mgr.(*manager).ReleaseForCache("cache1", 100, 1)
+
+	// Wait a bit to ensure no async metrics are emitted
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify NO metrics were emitted (no calls to UpdateGauge or IncCounter)
+	mockScope.AssertNotCalled(t, "UpdateGauge")
+	mockScope.AssertNotCalled(t, "IncCounter")
+}

common/dynamicconfig/dynamicproperties/constants.go

@@ -1828,12 +1828,6 @@ const (
 	// Default value: true
 	// Allowed filters: DomainID, WorkflowID
 	EnableReplicationTaskGeneration
-	// EnableReplicationBudgetManager enables the replication budget manager for cache capacity control
-	// KeyName: history.enableReplicationBudgetManager
-	// Value type: Bool
-	// Default value: false
-	// Allowed filters: N/A
-	EnableReplicationBudgetManager
 	// UseNewInitialFailoverVersion is a switch to issue a failover version based on the minFailoverVersion
 	// rather than the default initialFailoverVersion. USed as a per-domain migration switch
 	// KeyName: history.useNewInitialFailoverVersion
@@ -4442,11 +4436,6 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Description:  "EnableReplicationTaskGeneration is the flag to control replication generation",
 		DefaultValue: true,
 	},
-	EnableReplicationBudgetManager: {
-		KeyName:      "history.enableReplicationBudgetManager",
-		Description:  "EnableReplicationBudgetManager enables the replication budget manager for cache capacity control",
-		DefaultValue: false,
-	},
 	UseNewInitialFailoverVersion: {
 		KeyName:      "history.useNewInitialFailoverVersion",
 		Description:  "use the minInitialFailover version",

config/dynamicconfig/replication_simulation_budget_manager.yml

@@ -17,11 +17,6 @@ frontend.failoverCoolDown:
 history.ReplicationTaskProcessorStartWait: # default is 5s. repl task processor sleeps this much before processing received messages.
   - value: 10ms
 
-# Budget Manager Configuration - Enable the manager
-history.enableReplicationBudgetManager:
-  - value: true
-    constraints: {}
-
 # Set budget limits for testing
 # Max size in bytes - set to a moderate value to test capacity enforcement
 history.replicationBudgetManagerMaxSizeBytes:

service/history/config/config.go

@@ -191,7 +191,6 @@ type Config struct {
 	ReplicationBudgetManagerMaxSizeBytes     dynamicproperties.IntPropertyFn
 	ReplicationBudgetManagerMaxSizeCount     dynamicproperties.IntPropertyFn
 	ReplicationBudgetManagerSoftCapThreshold dynamicproperties.FloatPropertyFn
-	EnableReplicationBudgetManager           dynamicproperties.BoolPropertyFn
 
 	// System Limits
 	MaximumBufferedEventsBatch dynamicproperties.IntPropertyFn
@@ -491,7 +490,6 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, maxMessageSize int, i
 		ReplicationBudgetManagerMaxSizeBytes:     dc.GetIntProperty(dynamicproperties.ReplicationBudgetManagerMaxSizeBytes),
 		ReplicationBudgetManagerMaxSizeCount:     dc.GetIntProperty(dynamicproperties.ReplicationBudgetManagerMaxSizeCount),
 		ReplicationBudgetManagerSoftCapThreshold: dc.GetFloat64Property(dynamicproperties.ReplicationBudgetManagerSoftCapThreshold),
-		EnableReplicationBudgetManager:           dc.GetBoolProperty(dynamicproperties.EnableReplicationBudgetManager),
 
 		MaximumBufferedEventsBatch:      dc.GetIntProperty(dynamicproperties.MaximumBufferedEventsBatch),
 		MaximumSignalsPerExecution:      dc.GetIntPropertyFilteredByDomain(dynamicproperties.MaximumSignalsPerExecution),

service/history/config/config_test.go

@@ -170,7 +170,6 @@ func TestNewConfig(t *testing.T) {
 		"ReplicationBudgetManagerMaxSizeBytes":                 {dynamicproperties.ReplicationBudgetManagerMaxSizeBytes, 0},
 		"ReplicationBudgetManagerMaxSizeCount":                 {dynamicproperties.ReplicationBudgetManagerMaxSizeCount, 0},
 		"ReplicationBudgetManagerSoftCapThreshold":             {dynamicproperties.ReplicationBudgetManagerSoftCapThreshold, 1.0},
-		"EnableReplicationBudgetManager":                       {dynamicproperties.EnableReplicationBudgetManager, false},
 		"ExecutionMgrNumConns":                                 {dynamicproperties.ExecutionMgrNumConns, 57},
 		"HistoryMgrNumConns":                                   {dynamicproperties.HistoryMgrNumConns, 58},
 		"MaximumBufferedEventsBatch":                           {dynamicproperties.MaximumBufferedEventsBatch, 59},

service/history/handler/handler.go

@@ -137,18 +137,16 @@ func (h *handlerImpl) Start() {
 		h.config,
 	)
 
-	if h.config.EnableReplicationBudgetManager() {
-		h.replicationBudgetManager = cache.NewBudgetManager(
-			"replication-budget-manager",
-			h.config.ReplicationBudgetManagerMaxSizeBytes,
-			h.config.ReplicationBudgetManagerMaxSizeCount,
-			cache.AdmissionOptimistic,
-			0,
-			h.GetMetricsClient().Scope(metrics.ReplicatorCacheManagerScope),
-			h.GetLogger(),
-			h.config.ReplicationBudgetManagerSoftCapThreshold,
-		)
-	}
+	h.replicationBudgetManager = cache.NewBudgetManager(
+		"replication-budget-manager",
+		h.config.ReplicationBudgetManagerMaxSizeBytes,
+		h.config.ReplicationBudgetManagerMaxSizeCount,
+		cache.AdmissionOptimistic,
+		0,
+		h.GetMetricsClient().Scope(metrics.ReplicatorCacheManagerScope),
+		h.GetLogger(),
+		h.config.ReplicationBudgetManagerSoftCapThreshold,
+	)
 
 	h.controller = shard.NewShardController(
 		h.Resource,