Add an integration test for durable archival (#3732)

Author: MichaelSnowden

common/dynamicconfig/constants.go

@@ -524,6 +524,8 @@ const (
 	ArchivalProcessorRetryWarningLimit = "history.archivalProcessorRetryLimitWarning"
 	// ArchivalBackendMaxRPS is the maximum rate of requests per second to the archival backend
 	ArchivalBackendMaxRPS = "history.archivalBackendMaxRPS"
+	// DurableArchivalEnabled is the flag to enable durable archival
+	DurableArchivalEnabled = "history.durableArchivalEnabled"
 
 	// ReplicatorTaskBatchSize is batch size for ReplicatorProcessor
 	ReplicatorTaskBatchSize = "history.replicatorTaskBatchSize"

host/archival_test.go

@@ -68,9 +68,6 @@ type archivalSuite struct {
 }
 
 func (s *archivalSuite) SetupSuite() {
-	s.dynamicConfigOverrides = map[dynamicconfig.Key]interface{}{
-		dynamicconfig.RetentionTimerJitterDuration: time.Second,
-	}
 	s.setupSuite("testdata/integration_test_cluster.yaml")
 }
 
@@ -85,7 +82,30 @@ func (s *archivalSuite) SetupTest() {
 
 func TestArchivalSuite(t *testing.T) {
 	flag.Parse()
-	suite.Run(t, new(archivalSuite))
+	for _, c := range []struct {
+		Name                     string
+		DurableArchivalIsEnabled bool
+	}{
+		{
+			Name:                     "DurableArchivalIsDisabled",
+			DurableArchivalIsEnabled: false,
+		},
+		{
+			Name:                     "DurableArchivalIsEnabled",
+			DurableArchivalIsEnabled: true,
+		},
+	} {
+		c := c
+		t.Run(c.Name, func(t *testing.T) {
+			s := new(archivalSuite)
+			s.dynamicConfigOverrides = map[dynamicconfig.Key]interface{}{
+				dynamicconfig.RetentionTimerJitterDuration:  time.Second,
+				dynamicconfig.ArchivalProcessorArchiveDelay: time.Duration(0),
+				dynamicconfig.DurableArchivalEnabled:        c.DurableArchivalIsEnabled,
+			}
+			suite.Run(t, s)
+		})
+	}
 }
 
 func (s *archivalSuite) TestArchival_TimerQueueProcessor() {

service/history/configs/config.go

@@ -425,10 +425,7 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int32, isAdvancedVis
 		NumArchiveSystemWorkflows: dc.GetIntProperty(dynamicconfig.NumArchiveSystemWorkflows, 1000),
 		ArchiveRequestRPS:         dc.GetIntProperty(dynamicconfig.ArchiveRequestRPS, 300), // should be much smaller than frontend RPS
 		ArchiveSignalTimeout:      dc.GetDurationProperty(dynamicconfig.ArchiveSignalTimeout, 300*time.Millisecond),
-		DurableArchivalEnabled: func() bool {
-			// Always return false for now until durable archival is tested end-to-end
-			return false
-		},
+		DurableArchivalEnabled:    dc.GetBoolProperty(dynamicconfig.DurableArchivalEnabled, false),
 
 		BlobSizeLimitError:             dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitError, 2*1024*1024),
 		BlobSizeLimitWarn:              dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitWarn, 512*1024),

service/history/queueFactoryBase.go

@@ -109,6 +109,10 @@ var QueueModule = fx.Options(
 			Group:  QueueFactoryFxGroup,
 			Target: NewVisibilityQueueFactory,
 		},
+		fx.Annotated{
+			Group:  QueueFactoryFxGroup,
+			Target: NewArchivalQueueFactory,
+		},
 	),
 	fx.Invoke(QueueFactoryLifetimeHooks),
 )