Set up integration test for pending task alert of history queue v2 (#7196)

Shaddoll · web-flow · commit 23eede13efe9 · 2025-08-21T12:06:44.000-07:00
diff --git a/.github/workflows/ci-checks.yml b/.github/workflows/ci-checks.yml
@@ -133,6 +133,37 @@ jobs:
           name: go-cassandra-running-history-queue-v2-integration-coverage
           path: .build/coverage/*.out
 
+  golang-integration-test-with-cassandra-running-history-queue-v2-alert:
+    name: Golang integration test with running history queue v2 with alert
+    runs-on: ubuntu-latest
+    continue-on-error: true
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          submodules: true
+
+      - name: Setup Go environment
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.23.4'
+
+      - name: Run integration profile for cassandra running history queue v2 with alert
+        uses: nick-fields/retry@v3
+        with:
+          max_attempts: 2
+          timeout_minutes: 30
+          command: |
+            docker compose -f docker/github_actions/docker-compose.yml run integration-test-cassandra-queue-v2-alert bash -c "make .just-build && make cover_integration_profile"
+
+      - name: Upload coverage artifacts
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: go-cassandra-running-history-queue-v2-alert-integration-coverage
+          path: .build/coverage/*.out
+
 
   golang-integration-test-with-cassandra-and-elasticsearch-v7:
     name: Golang integration test with cassandra and elasticsearch v7
@@ -465,4 +496,4 @@ jobs:
           max_attempts: 2
           timeout_minutes: 30
           command: |
-            docker compose -f docker/github_actions/docker-compose.yml run integration-test-with-etcd bash -c "make .just-build && make integration_tests_etcd"
+            docker compose -f docker/github_actions/docker-compose.yml run integration-test-with-etcd bash -c "make .just-build && make integration_tests_etcd"
diff --git a/common/persistence/persistence-tests/persistenceTestBase.go b/common/persistence/persistence-tests/persistenceTestBase.go
@@ -156,8 +156,8 @@ func NewTestBaseWithNoSQL(t *testing.T, options *TestBaseOptions) *TestBase {
 		EnableShardIDMetrics:                     dynamicproperties.GetBoolPropertyFn(true),
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 	}
 	params := TestBaseParams{
 		DefaultTestCluster:    testCluster,
@@ -188,8 +188,8 @@ func NewTestBaseWithSQL(t *testing.T, options *TestBaseOptions) *TestBase {
 		EnableShardIDMetrics:                     dynamicproperties.GetBoolPropertyFn(true),
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 	}
 	params := TestBaseParams{
 		DefaultTestCluster:    testCluster,
diff --git a/docker/github_actions/docker-compose-local.yml b/docker/github_actions/docker-compose-local.yml
@@ -130,6 +130,37 @@ services:
         aliases:
           - integration-test
 
+  integration-test-cassandra-queue-v2-alert:
+    build:
+      context: ../../
+      dockerfile: ./docker/github_actions/Dockerfile${DOCKERFILE_SUFFIX}
+    command: make cover_integration_profile
+    environment:
+      - "CASSANDRA_HOST=cassandra"
+      - "CASSANDRA=1"
+      - "CASSANDRA_SEEDS=cassandra"
+      - "ES_SEEDS=elasticsearch"
+      - "KAFKA_SEEDS=kafka"
+      - "TEST_TAG=esintegration"
+      - "ENABLE_QUEUE_V2=true"
+      - "ENABLE_QUEUE_V2_ALERT=true"
+      - "V=1"
+    depends_on:
+      cassandra:
+        condition: service_healthy
+      elasticsearch:
+        condition: service_started
+      kafka:
+        condition: service_started
+    volumes:
+      - ../../:/cadence
+      - /cadence/.build/ # ensure we don't mount the build directory
+      - /cadence/.bin/ # ensure we don't mount the bin directory
+    networks:
+      services-network:
+        aliases:
+          - integration-test
+
   integration-test-mysql:
     build:
       context: ../../
diff --git a/docker/github_actions/docker-compose.yml b/docker/github_actions/docker-compose.yml
@@ -159,6 +159,32 @@ services:
         aliases:
           - integration-test
 
+  integration-test-cassandra-queue-v2-alert:
+    build:
+      context: ../../
+      dockerfile: ./docker/github_actions/Dockerfile${DOCKERFILE_SUFFIX}
+    environment:
+      - "CASSANDRA=1"
+      - "CASSANDRA_SEEDS=cassandra"
+      - "ES_SEEDS=elasticsearch"
+      - "KAFKA_SEEDS=kafka"
+      - "TEST_TAG=esintegration"
+      - "ENABLE_QUEUE_V2=true"
+      - "ENABLE_QUEUE_V2_ALERT=true"
+    depends_on:
+      cassandra:
+        condition: service_healthy
+      elasticsearch:
+        condition: service_started
+      kafka:
+        condition: service_started
+    volumes:
+      - ../../:/cadence
+    networks:
+      services-network:
+        aliases:
+          - integration-test
+
   integration-test-mysql:
     build:
       context: ../../
diff --git a/host/async_wf_test.go b/host/async_wf_test.go
@@ -99,8 +99,8 @@ func (s *AsyncWFIntegrationSuite) SetupSuite() {
 		EnableShardIDMetrics:                     dynamicproperties.GetBoolPropertyFn(true),
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 	}
 	params := pt.TestBaseParams{
 		DefaultTestCluster:    s.DefaultTestCluster,
diff --git a/host/integration_test.go b/host/integration_test.go
@@ -50,8 +50,12 @@ func TestIntegrationSuite(t *testing.T) {
 	flag.Parse()
 
 	configPath := "testdata/integration_test_cluster.yaml"
+	// TODO: remove this logic once we deprecate history queue v1
 	if os.Getenv("ENABLE_QUEUE_V2") == "true" {
 		configPath = "testdata/integration_queuev2_cluster.yaml"
+		if os.Getenv("ENABLE_QUEUE_V2_ALERT") == "true" {
+			configPath = "testdata/integration_queuev2_with_alert_cluster.yaml"
+		}
 	}
 	clusterConfig, err := GetTestClusterConfig(configPath)
 	if err != nil {
diff --git a/host/integrationbase.go b/host/integrationbase.go
@@ -116,8 +116,8 @@ func (s *IntegrationBase) setupSuite() {
 			EnableShardIDMetrics:                     dynamicproperties.GetBoolPropertyFn(true),
 			EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 			ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-			ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 			SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+			ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 		}
 		params := pt.TestBaseParams{
 			DefaultTestCluster:    s.DefaultTestCluster,
diff --git a/host/ndc/integration_test.go b/host/ndc/integration_test.go
@@ -102,8 +102,8 @@ func (s *NDCIntegrationTestSuite) SetupSuite() {
 		EnableCassandraAllConsistencyLevelDelete: dynamicproperties.GetBoolPropertyFn(true),
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 	}
 	params := pt.TestBaseParams{
 		DefaultTestCluster:    s.defaultTestCluster,
diff --git a/host/pinot_test.go b/host/pinot_test.go
@@ -114,8 +114,8 @@ func (s *PinotIntegrationSuite) SetupSuite() {
 		EnableShardIDMetrics:                     dynamicproperties.GetBoolPropertyFn(true),
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 	}
 	params := pt.TestBaseParams{
 		DefaultTestCluster:    s.DefaultTestCluster,
diff --git a/host/testdata/dynamicconfig/integration_queuev2_with_alert_test.yaml b/host/testdata/dynamicconfig/integration_queuev2_with_alert_test.yaml
@@ -0,0 +1,59 @@
+frontend.warmupDuration:
+- value: "1s"
+  constraints: {}
+history.enableTimerQueueV2:
+- value: true
+  constraints: {}
+history.enableTransferQueueV2:
+- value: true
+  constraints: {}
+history.shardUpdateMinInterval:
+- value: 3s
+  constraints: {}
+history.timerProcessorUpdateAckInterval:
+- value: 5s
+  constraints: {}
+history.timerProcessorUpdateAckIntervalJitterCoefficient:
+- value: 0
+  constraints: {}
+history.transferProcessorUpdateAckInterval:
+- value: 5s
+  constraints: {}
+history.transferProcessorUpdateAckIntervalJitterCoefficient:
+- value: 0
+  constraints: {}
+history.queueProcessorPollBackoffInterval:
+- value: 5s
+  constraints: {}
+history.virtualSliceForceAppendInterval:
+- value: 100ms
+  constraints: {}
+# Only Enable 1 level of split, which has been verified in simulation
+history.queueMaxVirtualQueueCount:
+- value: 2
+  constraints: {}
+# Enable task rate limiter so that the number of pending tasks increases
+history.taskSchedulerEnableRateLimiter:
+- value: true
+  constraints: {}
+history.taskSchedulerEnableRateLimiterShadowMode:
+- value: false
+  constraints: {}
+history.taskSchedulerGlobalDomainRPS:
+- value: 30
+  constraints: {}
+# Enable pending task queue alert for integration test
+history.enableTransferQueueV2PendingTaskCountAlert:
+- value: true
+  constraints: {}
+history.enableTimerQueueV2PendingTaskCountAlert:
+- value: true
+  constraints: {}
+# Set a low number to trigger queue pause with load from tests
+history.queueMaxPendingTaskCount:
+- value: 20
+  constraints: {}
+# Set a low number to trigger queue split with load from tests
+history.queueCriticalPendingTaskCount:
+- value: 18
+  constraints: {}
diff --git a/host/testdata/integration_queuev2_with_alert_cluster.yaml b/host/testdata/integration_queuev2_with_alert_cluster.yaml
@@ -0,0 +1,16 @@
+enablearchival: true
+clusterno: 0
+messagingclientconfig:
+  usemock: true
+historyconfig:
+  numhistoryshards: 4
+  numhistoryhosts: 1
+matchingconfig:
+  nummatchinghosts: 1
+workerconfig:
+  enablearchiver: true
+  enablereplicator: true
+  enableindexer: false
+dynamicclientconfig:
+  filepath: "testdata/dynamicconfig/integration_queuev2_with_alert_test.yaml"
+  pollInterval: "10s"
diff --git a/host/workflowidratelimit_test.go b/host/workflowidratelimit_test.go
@@ -76,8 +76,8 @@ func (s *WorkflowIDRateLimitIntegrationSuite) SetupSuite() {
 		EnableShardIDMetrics:                     dynamicproperties.GetBoolPropertyFn(true),
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 	}
 	params := pt.TestBaseParams{
 		DefaultTestCluster:    s.DefaultTestCluster,
diff --git a/host/workflowsidinternalratelimit_test.go b/host/workflowsidinternalratelimit_test.go
@@ -80,8 +80,8 @@ func (s *WorkflowIDInternalRateLimitIntegrationSuite) SetupSuite() {
 		EnableShardIDMetrics:                     dynamicproperties.GetBoolPropertyFn(true),
 		EnableHistoryTaskDualWriteMode:           dynamicproperties.GetBoolPropertyFn(true),
 		ReadNoSQLHistoryTaskFromDataBlob:         dynamicproperties.GetBoolPropertyFn(false),
-		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(false),
 		SerializationEncoding:                    dynamicproperties.GetStringPropertyFn(string(constants.EncodingTypeThriftRW)),
+		ReadNoSQLShardFromDataBlob:               dynamicproperties.GetBoolPropertyFn(true),
 	}
 	params := pt.TestBaseParams{
 		DefaultTestCluster:    s.DefaultTestCluster,
diff --git a/service/history/queuev2/virtual_queue.go b/service/history/queuev2/virtual_queue.go
@@ -70,7 +70,7 @@ type (
 	}
 
 	virtualQueueImpl struct {
-		options             *VirtualQueueOptions
+		queueOptions        *VirtualQueueOptions
 		processor           task.Processor
 		redispatcher        task.Redispatcher
 		logger              log.Logger
@@ -100,7 +100,7 @@ func NewVirtualQueue(
 	taskLoadRateLimiter quotas.Limiter,
 	monitor Monitor,
 	virtualSlices []VirtualSlice,
-	options *VirtualQueueOptions,
+	queueOptions *VirtualQueueOptions,
 ) VirtualQueue {
 	ctx, cancel := context.WithCancel(context.Background())
 
@@ -110,7 +110,7 @@ func NewVirtualQueue(
 	}
 
 	return &virtualQueueImpl{
-		options:             options,
+		queueOptions:        queueOptions,
 		processor:           processor,
 		redispatcher:        redispatcher,
 		logger:              logger,
@@ -339,10 +339,10 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {
 	}
 
 	pendingTaskCount := q.monitor.GetTotalPendingTaskCount()
-	maxTaskCount := q.options.MaxPendingTasksCount()
+	maxTaskCount := q.queueOptions.MaxPendingTasksCount()
 	if pendingTaskCount >= maxTaskCount {
 		q.logger.Warn("Too many pending tasks, pause loading tasks for a while", tag.PendingTaskCount(pendingTaskCount), tag.MaxTaskCount(maxTaskCount))
-		q.pauseController.Pause(q.options.PollBackoffInterval())
+		q.pauseController.Pause(q.queueOptions.PollBackoffInterval())
 	}
 
 	if q.pauseController.IsPaused() {
@@ -363,7 +363,7 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {
 		remainingSize = 1
 		q.logger.Error("unexpected error, virtual queue is not paused when pending task count exceeds max task cout limit", tag.PendingTaskCount(pendingTaskCount), tag.MaxTaskCount(maxTaskCount))
 	}
-	pageSize := min(q.options.PageSize(), remainingSize)
+	pageSize := min(q.queueOptions.PageSize(), remainingSize)
 	q.logger.Debug("get tasks from virtual queue", tag.PendingTaskCount(pendingTaskCount), tag.MaxTaskCount(maxTaskCount), tag.Counter(pageSize))
 	tasks, err := sliceToRead.GetTasks(q.ctx, pageSize)
 	if err != nil {
diff --git a/service/history/queuev2/virtual_queue_manager.go b/service/history/queuev2/virtual_queue_manager.go
@@ -68,7 +68,7 @@ type (
 		timeSource          clock.TimeSource
 		taskLoadRateLimiter quotas.Limiter
 		monitor             Monitor
-		options             *VirtualQueueManagerOptions
+		queueManagerOptions *VirtualQueueManagerOptions
 
 		sync.RWMutex
 		status               int32
@@ -89,7 +89,7 @@ func NewVirtualQueueManager(
 	timeSource clock.TimeSource,
 	taskLoadRateLimiter quotas.Limiter,
 	monitor Monitor,
-	options *VirtualQueueManagerOptions,
+	queueManagerOptions *VirtualQueueManagerOptions,
 	virtualQueueStates map[int64][]VirtualSliceState,
 ) VirtualQueueManager {
 	virtualQueues := make(map[int64]VirtualQueue)
@@ -100,9 +100,9 @@ func NewVirtualQueueManager(
 		}
 		var opts *VirtualQueueOptions
 		if queueID == rootQueueID {
-			opts = options.RootQueueOptions
+			opts = queueManagerOptions.RootQueueOptions
 		} else {
-			opts = options.NonRootQueueOptions
+			opts = queueManagerOptions.NonRootQueueOptions
 		}
 		virtualQueues[queueID] = NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, virtualSlices, opts)
 	}
@@ -116,15 +116,15 @@ func NewVirtualQueueManager(
 		timeSource:          timeSource,
 		taskLoadRateLimiter: taskLoadRateLimiter,
 		monitor:             monitor,
-		options:             options,
+		queueManagerOptions: queueManagerOptions,
 		status:              common.DaemonStatusInitialized,
 		virtualQueues:       virtualQueues,
 		createVirtualQueueFn: func(queueID int64, s ...VirtualSlice) VirtualQueue {
 			var opts *VirtualQueueOptions
 			if queueID == rootQueueID {
-				opts = options.RootQueueOptions
+				opts = queueManagerOptions.RootQueueOptions
 			} else {
-				opts = options.NonRootQueueOptions
+				opts = queueManagerOptions.NonRootQueueOptions
 			}
 			return NewVirtualQueue(processor, redispatcher, logger.WithTags(tag.VirtualQueueID(queueID)), metricsScope, timeSource, taskLoadRateLimiter, monitor, s, opts)
 		},
@@ -222,7 +222,7 @@ func (m *virtualQueueManagerImpl) appendOrMergeSlice(vq VirtualQueue, s VirtualS
 	now := m.timeSource.Now()
 	if now.After(m.nextForceNewSliceTime) {
 		vq.AppendSlices(s)
-		m.nextForceNewSliceTime = now.Add(m.options.VirtualSliceForceAppendInterval())
+		m.nextForceNewSliceTime = now.Add(m.queueManagerOptions.VirtualSliceForceAppendInterval())
 		return
 	}
 	vq.MergeSlices(s)
diff --git a/service/history/queuev2/virtual_queue_manager_test.go b/service/history/queuev2/virtual_queue_manager_test.go
diff --git a/simulation/history/history_simulation_test.go b/simulation/history/history_simulation_test.go
