Unload shard if loaded task ID is greater than max transfer ID (#2361)

* Unload shard if loaded task ID is greater than max transfer ID

* Update max read level for timer task ID as well

service/history/shard/context.go

@@ -128,5 +128,7 @@ type (
 		GetSearchAttributesProvider() searchattribute.Provider
 		GetSearchAttributesMapper() searchattribute.Mapper
 		GetArchivalMetadata() archiver.ArchivalMetadata
+
+		Unload()
 	}
 )

service/history/shard/context_impl.go

@@ -1088,7 +1088,8 @@ func (s *ContextImpl) allocateTaskIDsLocked(
 	return s.allocateTimerIDsLocked(
 		namespaceEntry,
 		workflowID,
-		timerTasks)
+		timerTasks,
+		transferMaxReadLevel)
 }
 
 func (s *ContextImpl) allocateTransferIDsLocked(
@@ -1115,6 +1116,7 @@ func (s *ContextImpl) allocateTimerIDsLocked(
 	namespaceEntry *namespace.Namespace,
 	workflowID string,
 	timerTasks []tasks.Task,
+	transferMaxReadLevel *int64,
 ) error {
 
 	// assign IDs for the timer tasks. They need to be assigned under shard lock.
@@ -1145,6 +1147,7 @@ func (s *ContextImpl) allocateTimerIDsLocked(
 			return err
 		}
 		task.SetTaskID(seqNum)
+		*transferMaxReadLevel = seqNum
 		visibilityTs := task.GetVisibilityTime()
 		s.contextTaggedLogger.Debug("Assigning new timer",
 			tag.Timestamp(visibilityTs), tag.TaskID(task.GetTaskID()), tag.AckLevel(s.shardInfo.TimerAckLevelTime))
@@ -1269,6 +1272,10 @@ func (s *ContextImpl) start() {
 	s.transitionLocked(contextRequestAcquire{})
 }
 
+func (s *ContextImpl) Unload() {
+	s.stop()
+}
+
 // stop should only be called by the controller.
 func (s *ContextImpl) stop() {
 	s.wLock()

service/history/taskProcessor.go

@@ -168,6 +168,12 @@ func (t *taskProcessor) taskWorker(
 			if !ok {
 				return
 			}
+			if task.GetTaskID() > t.shard.GetTransferMaxReadLevel() {
+				// this could happen if we lost ownership and was not aware of it.
+				// unload shard
+				t.shard.Unload()
+				return
+			}
 			t.processTaskAndAck(notificationChan, task)
 		}
 	}