Add handover namespaces to GetReplicationStatus response (temporalio#…

…2248)

* Add handover namespaces to GetReplicationStatus response

proto/internal/temporal/server/api/historyservice/v1/request_response.proto

@@ -526,6 +526,12 @@ message ShardReplicationStatus {
     // Local time on this shard
     google.protobuf.Timestamp shard_local_time = 3 [(gogoproto.stdtime) = true];
     map<string, ShardReplicationStatusPerCluster> remote_clusters = 4;
+    map<string, HandoverNamespaceInfo> handover_namespaces = 5;
+}
+
+message HandoverNamespaceInfo {
+    // max replication task id when namespace transition to Handover state
+    int64 handover_replication_task_id = 1;
 }
 
 message ShardReplicationStatusPerCluster {

service/history/historyEngine.go

@@ -3288,11 +3288,12 @@ func (e *historyEngineImpl) GetReplicationStatus(
 		resp.MaxReplicationTaskId = *e.replicatorProcessor.maxTaskID
 	}
 
-	remoteClusters, err := e.shard.GetRemoteClusterAckInfo(request.RemoteClusters)
+	remoteClusters, handoverNamespaces, err := e.shard.GetReplicationStatus(request.RemoteClusters)
 	if err != nil {
 		return nil, err
 	}
 	resp.RemoteClusters = remoteClusters
+	resp.HandoverNamespaces = handoverNamespaces
 	return resp, nil
 }
 

service/history/shard/context.go

@@ -74,7 +74,7 @@ type (
 		GetLastUpdatedTime() time.Time
 		GetTimerMaxReadLevel(cluster string) time.Time
 
-		GetRemoteClusterAckInfo(cluster []string) (map[string]*historyservice.ShardReplicationStatusPerCluster, error)
+		GetReplicationStatus(cluster []string) (map[string]*historyservice.ShardReplicationStatusPerCluster, map[string]*historyservice.HandoverNamespaceInfo, error)
 
 		GetTransferAckLevel() int64
 		UpdateTransferAckLevel(ackLevel int64) error

service/history/shard/context_impl.go

@@ -1488,30 +1488,38 @@ func (s *ContextImpl) loadShardMetadata(ownershipChanged *bool) error {
 	return nil
 }
 
-func (s *ContextImpl) GetRemoteClusterAckInfo(cluster []string) (map[string]*historyservice.ShardReplicationStatusPerCluster, error) {
-	resp := make(map[string]*historyservice.ShardReplicationStatusPerCluster)
+func (s *ContextImpl) GetReplicationStatus(cluster []string) (map[string]*historyservice.ShardReplicationStatusPerCluster, map[string]*historyservice.HandoverNamespaceInfo, error) {
+	remoteClusters := make(map[string]*historyservice.ShardReplicationStatusPerCluster)
+	handoverNamespaces := make(map[string]*historyservice.HandoverNamespaceInfo)
 	s.rLock()
 	defer s.rUnlock()
+
 	if len(cluster) == 0 {
 		// remote acked info for all known remote clusters
 		for k, v := range s.remoteClusterInfos {
-			resp[k] = &historyservice.ShardReplicationStatusPerCluster{
+			remoteClusters[k] = &historyservice.ShardReplicationStatusPerCluster{
 				AckedTaskId:             v.AckedReplicationTaskID,
 				AckedTaskVisibilityTime: timestamp.TimePtr(v.AckedReplicationTimestamp),
 			}
 		}
 	} else {
 		for _, k := range cluster {
 			if v, ok := s.remoteClusterInfos[k]; ok {
-				resp[k] = &historyservice.ShardReplicationStatusPerCluster{
+				remoteClusters[k] = &historyservice.ShardReplicationStatusPerCluster{
 					AckedTaskId:             v.AckedReplicationTaskID,
 					AckedTaskVisibilityTime: timestamp.TimePtr(v.AckedReplicationTimestamp),
 				}
 			}
 		}
 	}
 
-	return resp, nil
+	for k, v := range s.handoverNamespaces {
+		handoverNamespaces[k] = &historyservice.HandoverNamespaceInfo{
+			HandoverReplicationTaskId: v.MaxReplicationTaskID,
+		}
+	}
+
+	return remoteClusters, handoverNamespaces, nil
 }
 
 func (s *ContextImpl) getRemoteClusterInfoLocked(clusterName string) *remoteClusterInfo {