Skip to content

Commit

Permalink
Conditionally register archival category
Browse files Browse the repository at this point in the history
  • Loading branch information
MichaelSnowden committed Jan 28, 2023
1 parent 6e732e7 commit 09c9c42
Show file tree
Hide file tree
Showing 3 changed files with 25 additions and 1 deletion.
4 changes: 4 additions & 0 deletions service/history/queueFactoryBase.go
Expand Up @@ -42,6 +42,7 @@ import (
"go.temporal.io/server/service/history/configs"
"go.temporal.io/server/service/history/queues"
"go.temporal.io/server/service/history/shard"
"go.temporal.io/server/service/history/tasks"
wcache "go.temporal.io/server/service/history/workflow/cache"
)

Expand Down Expand Up @@ -136,8 +137,11 @@ func getQueueFactories(
queueFactorySet.TimerQueueFactory,
queueFactorySet.VisibilityQueueFactory,
}
c := tasks.CategoryArchival
tasks.RemoveCategory(c.ID())
if archivalMetadata.GetHistoryConfig().StaticClusterState() == archiver.ArchivalEnabled || archivalMetadata.GetVisibilityConfig().StaticClusterState() == archiver.ArchivalEnabled {
factories = append(factories, queueFactorySet.ArchivalQueueFactory)
tasks.UpsertCategory(c)
}
return factories
}
Expand Down
6 changes: 5 additions & 1 deletion service/history/queue_factory_base_test.go
Expand Up @@ -28,6 +28,7 @@ import (
"testing"

"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/fx"

Expand All @@ -45,6 +46,7 @@ import (
"go.temporal.io/server/common/sdk"
"go.temporal.io/server/service/history/archival"
"go.temporal.io/server/service/history/configs"
"go.temporal.io/server/service/history/tasks"
"go.temporal.io/server/service/history/workflow"
"go.temporal.io/server/service/worker/archiver"
)
Expand Down Expand Up @@ -93,7 +95,7 @@ type moduleTestCase struct {

// Run runs the test case.
func (c *moduleTestCase) Run(t *testing.T) {
t.Parallel()

controller := gomock.NewController(t)
dependencies := getModuleDependencies(controller, c)
var factories []QueueFactory
Expand Down Expand Up @@ -135,8 +137,10 @@ func (c *moduleTestCase) Run(t *testing.T) {
require.NotNil(t, viq)
if c.ExpectArchivalQueue {
require.NotNil(t, aq)
assert.Contains(t, tasks.GetCategories(), tasks.CategoryIDArchival)
} else {
require.Nil(t, aq)
assert.NotContains(t, tasks.GetCategories(), tasks.CategoryIDArchival)
}
}

Expand Down
16 changes: 16 additions & 0 deletions service/history/tasks/category.go
Expand Up @@ -145,6 +145,22 @@ func GetCategories() map[int32]Category {
return maps.Clone(categories.m)
}

// UpsertCategory registers or updates a Category.
// This should only be used for testing.
func UpsertCategory(category Category) {
categories.Lock()
defer categories.Unlock()
categories.m[category.ID()] = category
}

// RemoveCategory removes a registered Category.
// This should only be used for testing.
func RemoveCategory(id int32) {
categories.Lock()
defer categories.Unlock()
delete(categories.m, id)
}

// GetCategoryByID returns a registered Category with the same ID
func GetCategoryByID(id int32) (Category, bool) {
categories.RLock()
Expand Down

0 comments on commit 09c9c42

Please sign in to comment.