Consistent query bug fixes (cadence-workflow#2908)

MartinGomes · Dec 13, 2019 · 39fabad · 39fabad
1 parent 1e63e93
commit 39fabad
Show file tree

Hide file tree

Showing 6 changed files with 36 additions and 13 deletions.
diff --git a/common/client/versionChecker.go b/common/client/versionChecker.go
@@ -40,11 +40,11 @@ const (
 	CLI = "cli"
 
 	// SupportedGoSDKVersion indicates the highest go sdk version server will accept requests from
-	SupportedGoSDKVersion = "1.4.0"
+	SupportedGoSDKVersion = "1.5.0"
 	// SupportedJavaSDKVersion indicates the highest java sdk version server will accept requests from
-	SupportedJavaSDKVersion = "1.4.0"
+	SupportedJavaSDKVersion = "1.5.0"
 	// SupportedCLIVersion indicates the highest cli version server will accept requests from
-	SupportedCLIVersion = "1.4.0"
+	SupportedCLIVersion = "1.5.0"
 
 	// GoWorkerStickyQueryVersion indicates the minimum client version of go worker which supports StickyQuery
 	GoWorkerStickyQueryVersion = "1.0.0"

diff --git a/config/dynamicconfig/development.yaml b/config/dynamicconfig/development.yaml
@@ -3,4 +3,7 @@ frontend.enableClientVersionCheck:
   constraints: {}
 system.minRetentionDays:
 - value: 0
-  constraints: {}
+  constraints: {}
+history.EnableConsistentQueryByDomain:
+  - value: true
+    constrains: {}
diff --git a/service/frontend/workflowHandler.go b/service/frontend/workflowHandler.go
@@ -3116,7 +3116,7 @@ func (wh *WorkflowHandler) getDefaultScope(scope int) metrics.Scope {
 	return wh.GetMetricsClient().Scope(scope).Tagged(metrics.DomainUnknownTag())
 }
 
-func frontendInternalServiceError(fmtStr string, args... interface{}) error{
+func frontendInternalServiceError(fmtStr string, args ...interface{}) error {
 	// NOTE: For internal error, we can't return thrift error from cadence-frontend.
 	// Because in uber internal metrics, thrift errors are counted as user errors.
 	return fmt.Errorf(fmtStr, args...)

diff --git a/service/history/decisionHandler.go b/service/history/decisionHandler.go
@@ -550,7 +550,14 @@ Update_History_Loop:
 			return nil, updateErr
 		}
 
-		handler.handleBufferedQueries(msBuilder, clientImpl, clientFeatureVersion, req.GetCompleteRequest().GetQueryResults(), createNewDecisionTask, domainEntry)
+		handler.handleBufferedQueries(
+			msBuilder,
+			clientImpl,
+			clientFeatureVersion,
+			req.GetCompleteRequest().GetQueryResults(),
+			createNewDecisionTask,
+			domainEntry,
+			decisionHeartbeating)
 
 		if decisionHeartbeatTimeout {
 			// at this point, update is successful, but we still return an error to client so that the worker will give up this workflow
@@ -639,6 +646,7 @@ func (handler *decisionHandlerImpl) handleBufferedQueries(
 	queryResults map[string]*workflow.WorkflowQueryResult,
 	createNewDecisionTask bool,
 	domainEntry *cache.DomainCacheEntry,
+	decisionHeartbeating bool,
 ) {
 	queryRegistry := msBuilder.GetQueryRegistry()
 	if !queryRegistry.hasBufferedQuery() {
@@ -658,7 +666,7 @@ func (handler *decisionHandlerImpl) handleBufferedQueries(
 		scope.IncCounter(metrics.WorkerNotSupportsConsistentQueryCount)
 		failedTerminationState := &queryTerminationState{
 			queryTerminationType: queryTerminationTypeFailed,
-			failure:              versionErr,
+			failure:              &workflow.BadRequestError{Message: versionErr.Error()},
 		}
 		buffered := queryRegistry.getBufferedIDs()
 		handler.logger.Info(
@@ -682,6 +690,12 @@ func (handler *decisionHandlerImpl) handleBufferedQueries(
 		return
 	}
 
+	// if its a heartbeat decision it means local activities may still be running on the worker
+	// which were started by an external event which happened before the query
+	if decisionHeartbeating {
+		return
+	}
+
 	sizeLimitError := handler.config.BlobSizeLimitError(domain)
 	sizeLimitWarn := handler.config.BlobSizeLimitWarn(domain)
 

diff --git a/service/history/decisionHandler_test.go b/service/history/decisionHandler_test.go
@@ -81,21 +81,28 @@ func (s *DecisionHandlerSuite) TearDownTest() {
 
 func (s *DecisionHandlerSuite) TestHandleBufferedQueries_ClientNotSupports() {
 	s.assertQueryCounts(s.queryRegistry, 10, 0, 0, 0)
-	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, "0.0.0", nil, false, testGlobalDomainEntry)
+	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, "0.0.0", nil, false, testGlobalDomainEntry, false)
 	s.assertQueryCounts(s.queryRegistry, 0, 0, 0, 10)
 }
 
+func (s *DecisionHandlerSuite) TestHandleBufferedQueries_HeartbeatDecision() {
+	s.assertQueryCounts(s.queryRegistry, 10, 0, 0, 0)
+	queryResults := s.constructQueryResults(s.queryRegistry.getBufferedIDs()[0:5], 10)
+	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, client.GoWorkerConsistentQueryVersion, queryResults, false, testGlobalDomainEntry, true)
+	s.assertQueryCounts(s.queryRegistry, 10, 0, 0, 0)
+}
+
 func (s *DecisionHandlerSuite) TestHandleBufferedQueries_NewDecisionTask() {
 	s.assertQueryCounts(s.queryRegistry, 10, 0, 0, 0)
 	queryResults := s.constructQueryResults(s.queryRegistry.getBufferedIDs()[0:5], 10)
-	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, client.GoWorkerConsistentQueryVersion, queryResults, true, testGlobalDomainEntry)
+	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, client.GoWorkerConsistentQueryVersion, queryResults, true, testGlobalDomainEntry, false)
 	s.assertQueryCounts(s.queryRegistry, 5, 5, 0, 0)
 }
 
 func (s *DecisionHandlerSuite) TestHandleBufferedQueries_NoNewDecisionTask() {
 	s.assertQueryCounts(s.queryRegistry, 10, 0, 0, 0)
 	queryResults := s.constructQueryResults(s.queryRegistry.getBufferedIDs()[0:5], 10)
-	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, client.GoWorkerConsistentQueryVersion, queryResults, false, testGlobalDomainEntry)
+	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, client.GoWorkerConsistentQueryVersion, queryResults, false, testGlobalDomainEntry, false)
 	s.assertQueryCounts(s.queryRegistry, 0, 5, 5, 0)
 }
 
@@ -107,7 +114,7 @@ func (s *DecisionHandlerSuite) TestHandleBufferedQueries_QueryTooLarge() {
 	for k, v := range largeQueryResults {
 		queryResults[k] = v
 	}
-	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, client.GoWorkerConsistentQueryVersion, queryResults, false, testGlobalDomainEntry)
+	s.decisionHandler.handleBufferedQueries(s.mockMutableState, client.GoSDK, client.GoWorkerConsistentQueryVersion, queryResults, false, testGlobalDomainEntry, false)
 	s.assertQueryCounts(s.queryRegistry, 0, 5, 0, 5)
 }
 

diff --git a/service/history/historyEngine.go b/service/history/historyEngine.go
@@ -175,7 +175,7 @@ var (
 	// ErrEventsAterWorkflowFinish is the error indicating server error trying to write events after workflow finish event
 	ErrEventsAterWorkflowFinish = &workflow.InternalServiceError{Message: "error validating last event being workflow finish event"}
 	// ErrQueryEnteredInvalidState is error indicating query entered invalid state
-	ErrQueryEnteredInvalidState = &workflow.InternalServiceError{Message: "query entered invalid state, this should be impossible"}
+	ErrQueryEnteredInvalidState = &workflow.BadRequestError{Message: "query entered invalid state, this should be impossible"}
 	// ErrQueryWorkflowBeforeFirstDecision is error indicating that query was attempted before first decision task completed
 	ErrQueryWorkflowBeforeFirstDecision = &workflow.BadRequestError{Message: "workflow must handle at least one decision task before it can be queried"}
 	// ErrConsistentQueryNotEnabled is error indicating that consistent query was requested but either cluster or domain does not enable consistent query
@@ -1002,7 +1002,6 @@ func (e *historyEngineImpl) queryDirectlyThroughMatching(
 func (e *historyEngineImpl) getMutableState(
 	ctx ctx.Context,
 	domainID string,
-
 	execution workflow.WorkflowExecution,
 ) (retResp *h.GetMutableStateResponse, retError error) {