merge main

.buildkite/pipeline-master.yml

@@ -304,5 +304,5 @@ steps:
                         name: ubercadence-dockerhub
                         key: password
       - docker-login#v2.0.1:
-          username: ubercadence
+          username: jht305
           password-env: DOCKER_LOGIN_PASSWORD

common/definition/indexedKeys_test.go

@@ -0,0 +1,44 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package definition
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsSystemBoolKey(t *testing.T) {
+	tests := []struct {
+		key      string
+		expected bool
+	}{
+		{"IsCron", true},
+		{"StartTime", false},
+	}
+
+	for _, test := range tests {
+		actualResult := IsSystemBoolKey(test.key)
+		assert.Equal(t, test.expected, actualResult)
+	}
+}

common/dynamicconfig/constants.go

@@ -1672,6 +1672,13 @@ const (
 	// Default value: false
 	// Allowed filters: DomainID
 	MatchingEnableTaskInfoLogByDomainID
+	// MatchingEnableTasklistGuardAgainstOwnershipShardLoss
+	// enables guards to prevent tasklists from processing if there is any detection that the host
+	// no longer is active or owns the shard
+	// KeyName: matching.enableTasklistGuardAgainstOwnershipLoss
+	// Value type: Bool
+	// Default value: false
+	MatchingEnableTasklistGuardAgainstOwnershipShardLoss
 
 	// key for history
 
@@ -2859,6 +2866,13 @@ const (
 	// Allowed filters: domainName, taskListName, taskListType
 	LocalPollWaitTime
 
+	// LocalTaskWaitTime is the wait time for a task to wait before considering task forwarding
+	// KeyName: matching.localTaskWaitTime
+	// Value type: Duration
+	// Default value: 10ms
+	// Allowed filters: domainName, taskListName, taskListType
+	LocalTaskWaitTime
+
 	// LastDurationKey must be the last one in this const group
 	LastDurationKey
 )
@@ -4109,6 +4123,11 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Description:  "MatchingEnableTaskInfoLogByDomainID is enables info level logs for decision/activity task based on the request domainID",
 		DefaultValue: false,
 	},
+	MatchingEnableTasklistGuardAgainstOwnershipShardLoss: {
+		KeyName:      "matching.enableTasklistGuardAgainstOwnershipLoss",
+		Description:  "allows guards to ensure that tasklists don't continue processing if there's signal that they've lost ownership",
+		DefaultValue: false,
+	},
 	EventsCacheGlobalEnable: {
 		KeyName:      "history.eventsCacheGlobalEnable",
 		Description:  "EventsCacheGlobalEnable is enables global cache over all history shards",
@@ -5148,6 +5167,12 @@ var DurationKeys = map[DurationKey]DynamicDuration{
 		Description:  "LocalPollWaitTime is the time a poller waits before considering request forwarding.",
 		DefaultValue: time.Millisecond * 10,
 	},
+	LocalTaskWaitTime: {
+		KeyName:      "matching.localTaskWaitTime",
+		Filters:      []Filter{DomainName, TaskListName, TaskType},
+		Description:  "LocalTaskWaitTime is the time a task waits for a poller to arrive before considering task forwarding",
+		DefaultValue: time.Millisecond * 10,
+	},
 }
 
 var MapKeys = map[MapKey]DynamicMap{

common/errors/taskListNotOwnedByHostError.go

@@ -24,21 +24,21 @@ package errors
 
 import "fmt"
 
-var _ error = &TaskListNotOwnnedByHostError{}
+var _ error = &TaskListNotOwnedByHostError{}
 
-type TaskListNotOwnnedByHostError struct {
+type TaskListNotOwnedByHostError struct {
 	OwnedByIdentity string
 	MyIdentity      string
 	TasklistName    string
 }
 
-func (m *TaskListNotOwnnedByHostError) Error() string {
+func (m *TaskListNotOwnedByHostError) Error() string {
 	return fmt.Sprintf("task list is not owned by this host: OwnedBy: %s, Me: %s, Tasklist: %s",
 		m.OwnedByIdentity, m.MyIdentity, m.TasklistName)
 }
 
-func NewTaskListNotOwnnedByHostError(ownedByIdentity string, myIdentity string, tasklistName string) *TaskListNotOwnnedByHostError {
-	return &TaskListNotOwnnedByHostError{
+func NewTaskListNotOwnedByHostError(ownedByIdentity string, myIdentity string, tasklistName string) *TaskListNotOwnedByHostError {
+	return &TaskListNotOwnedByHostError{
 		OwnedByIdentity: ownedByIdentity,
 		MyIdentity:      myIdentity,
 		TasklistName:    tasklistName,

common/log/tag/tags.go

@@ -956,6 +956,12 @@ func VisibilityQuery(query string) Tag {
 	return newStringTag("visibility-query", query)
 }
 
+// MembershipChangeEvent is a predefined tag for when logging hashring change events,
+// expected to be of type membership.ChangeEvent
+func MembershipChangeEvent(event interface{}) Tag {
+	return newPredefinedDynamicTag("membership-change-event", event)
+}
+
 // Dynamic Uses reflection based logging for arbitrary values
 // for not very performant logging
 func Dynamic(key string, v interface{}) Tag {

common/membership/resolver.go

@@ -25,6 +25,7 @@ package membership
 
 import (
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	"github.com/uber/cadence/common"
@@ -84,6 +85,7 @@ type MultiringResolver struct {
 	status  int32
 
 	provider PeerProvider
+	mu       sync.Mutex
 	rings    map[string]*ring
 }
 
@@ -110,6 +112,7 @@ func NewMultiringResolver(
 		provider: provider,
 		rings:    make(map[string]*ring),
 		metrics:  metricsClient,
+		mu:       sync.Mutex{},
 	}
 
 	for _, s := range services {
@@ -130,6 +133,8 @@ func (rpo *MultiringResolver) Start() {
 
 	rpo.provider.Start()
 
+	rpo.mu.Lock()
+	defer rpo.mu.Unlock()
 	for _, ring := range rpo.rings {
 		ring.Start()
 	}
@@ -145,6 +150,8 @@ func (rpo *MultiringResolver) Stop() {
 		return
 	}
 
+	rpo.mu.Lock()
+	defer rpo.mu.Unlock()
 	for _, ring := range rpo.rings {
 		ring.Stop()
 	}
@@ -163,6 +170,8 @@ func (rpo *MultiringResolver) EvictSelf() error {
 }
 
 func (rpo *MultiringResolver) getRing(service string) (*ring, error) {
+	rpo.mu.Lock()
+	defer rpo.mu.Unlock()
 	ring, found := rpo.rings[service]
 	if !found {
 		return nil, fmt.Errorf("service %q is not tracked by Resolver", service)

common/pinot/pinotQueryValidator_test.go

@@ -296,6 +296,11 @@ func TestValidateQuery(t *testing.T) {
 			validated: "",
 			err:       "invalid bool value in pinot_query_validator: 1",
 		},
+		"case21-5: test bool value- when it is not SQLBool and SQLVAl": {
+			query:     "IsCron = abc",
+			validated: "",
+			err:       "failed to process a bool key to SQLVal: &{<nil> abc { }}",
+		},
 	}
 
 	for name, test := range tests {