Report acked replication task timestamp (#2173)

Report back acked replication task timestamp So source cluster could know how far the remote cluster is lagging behind on replication queue.
2026-08-30 18:41:49 -07:00 · 2021-11-15 10:58:36 -08:00
parent 170bde6aca
commit 43ac47ffb4
14 changed files with 342 additions and 166 deletions
--- a/api/replication/v1/message.pb.go
+++ b/api/replication/v1/message.pb.go
@@ -71,7 +71,8 @@ type ReplicationTask struct {
 	//	*ReplicationTask_SyncActivityTaskAttributes
 	//	*ReplicationTask_HistoryMetadataTaskAttributes
 	//	*ReplicationTask_HistoryTaskV2Attributes
-	Attributes isReplicationTask_Attributes `protobuf_oneof:"attributes"`
+	Attributes     isReplicationTask_Attributes `protobuf_oneof:"attributes"`
+	VisibilityTime *time.Time                   `protobuf:"bytes,9,opt,name=visibility_time,json=visibilityTime,proto3,stdtime" json:"visibility_time,omitempty"`
 }

 func (m *ReplicationTask) Reset()      { *m = ReplicationTask{} }
@@ -202,6 +203,13 @@ func (m *ReplicationTask) GetHistoryTaskV2Attributes() *HistoryTaskV2Attributes
 	return nil
 }

+func (m *ReplicationTask) GetVisibilityTime() *time.Time {
+	if m != nil {
+		return m.VisibilityTime
+	}
+	return nil
+}
+
 // XXX_OneofWrappers is for the internal use of the proto package.
 func (*ReplicationTask) XXX_OneofWrappers() []interface{} {
 	return []interface{}{
@@ -221,6 +229,8 @@ type ReplicationToken struct {
 	// lastProcessedMessageId is the last messageId that is processed on the passive side.
 	// This can be different than lastRetrievedMessageId if passive side supports prefetching messages.
 	LastProcessedMessageId int64 `protobuf:"varint,3,opt,name=last_processed_message_id,json=lastProcessedMessageId,proto3" json:"last_processed_message_id,omitempty"`
+	// The VisibilityTime of last processed ReplicationTask
+	LastProcessedVisibilityTime *time.Time `protobuf:"bytes,4,opt,name=last_processed_visibility_time,json=lastProcessedVisibilityTime,proto3,stdtime" json:"last_processed_visibility_time,omitempty"`
 }

 func (m *ReplicationToken) Reset()      { *m = ReplicationToken{} }
@@ -276,6 +286,13 @@ func (m *ReplicationToken) GetLastProcessedMessageId() int64 {
 	return 0
 }

+func (m *ReplicationToken) GetLastProcessedVisibilityTime() *time.Time {
+	if m != nil {
+		return m.LastProcessedVisibilityTime
+	}
+	return nil
+}
+
 type SyncShardStatus struct {
 	StatusTime *time.Time `protobuf:"bytes,1,opt,name=status_time,json=statusTime,proto3,stdtime" json:"status_time,omitempty"`
 }
@@ -1093,101 +1110,104 @@ func init() {
 }

 var fileDescriptor_edd9fae2af6b0532 = []byte{
-	// 1503 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x58, 0x4f, 0x6f, 0x1a, 0x47,
-	0x1b, 0xf7, 0x02, 0x86, 0x65, 0xf8, 0x9b, 0xf1, 0xeb, 0xd7, 0x18, 0xc9, 0xc4, 0x46, 0xc9, 0x1b,
-	0xe7, 0x55, 0xb5, 0xc4, 0xf8, 0xd0, 0x26, 0xa9, 0x2a, 0xd9, 0x69, 0x53, 0x63, 0x29, 0x69, 0xb4,
-	0xb1, 0x12, 0xa9, 0x97, 0xed, 0x98, 0x1d, 0x60, 0x65, 0xd8, 0x45, 0x33, 0x03, 0x2e, 0x3d, 0x55,
-	0xea, 0xa1, 0x97, 0x56, 0xca, 0x77, 0x48, 0x55, 0xf5, 0xd4, 0xcf, 0x91, 0x63, 0x2e, 0x95, 0xd2,
-	0x53, 0x1b, 0xa7, 0x87, 0x1e, 0xf3, 0x11, 0xaa, 0xf9, 0xb3, 0xb0, 0xcb, 0x02, 0xa1, 0xa9, 0x72,
-	0xea, 0x8d, 0x79, 0xfe, 0xfc, 0x9e, 0x99, 0x67, 0x9e, 0xe7, 0xf9, 0xcd, 0x02, 0x6e, 0x30, 0xdc,
-	0xeb, 0x7b, 0x04, 0x75, 0x6b, 0x14, 0x93, 0x21, 0x26, 0x35, 0xd4, 0x77, 0x6a, 0x04, 0xf7, 0xbb,
-	0x4e, 0x13, 0x31, 0xc7, 0x73, 0x6b, 0xc3, 0xbd, 0x5a, 0x0f, 0x53, 0x8a, 0xda, 0xd8, 0xe8, 0x13,
-	0x8f, 0x79, 0xb0, 0xea, 0x7b, 0x18, 0xd2, 0xc3, 0x40, 0x7d, 0xc7, 0x08, 0x78, 0x18, 0xc3, 0xbd,
-	0xf2, 0xe5, 0xb6, 0xe7, 0xb5, 0xbb, 0xb8, 0x26, 0x3c, 0x4e, 0x07, 0xad, 0x1a, 0x73, 0x7a, 0x98,
-	0x32, 0xd4, 0xeb, 0x4b, 0x90, 0xf2, 0x8e, 0x8d, 0xfb, 0xd8, 0xb5, 0xb1, 0xdb, 0x74, 0x30, 0xad,
-	0xb5, 0xbd, 0xb6, 0x27, 0xe4, 0xe2, 0x97, 0x32, 0x31, 0x66, 0xed, 0x0c, 0xbb, 0x83, 0x1e, 0xe5,
-	0x7b, 0x0a, 0x06, 0x94, 0xf6, 0xd7, 0x16, 0xda, 0x33, 0x44, 0xcf, 0x94, 0xe1, 0x7b, 0xb3, 0x0c,
-	0x3b, 0x0e, 0x65, 0x1e, 0x19, 0x45, 0x8e, 0x5b, 0xbe, 0x32, 0xb6, 0xe6, 0x66, 0x4d, 0xaf, 0xd7,
-	0x9b, 0x91, 0x94, 0x40, 0x70, 0x6e, 0xe5, 0xa2, 0x1e, 0xa6, 0x7d, 0xd4, 0xc4, 0x51, 0xc3, 0xeb,
-	0x21, 0xc3, 0x45, 0x89, 0x2e, 0x5f, 0x0d, 0x99, 0xce, 0xdd, 0x60, 0xd8, 0xac, 0x85, 0x9c, 0xee,
-	0x80, 0x44, 0x03, 0x57, 0x7f, 0x4c, 0x81, 0x82, 0x39, 0x09, 0x77, 0x82, 0xe8, 0x19, 0xbc, 0x0f,
-	0xd2, 0x3c, 0x2f, 0x16, 0x1b, 0xf5, 0x71, 0x49, 0xdb, 0xd6, 0x76, 0xf3, 0xf5, 0x3d, 0x63, 0xd6,
-	0xf5, 0x8a, 0x34, 0x1a, 0xc3, 0x3d, 0x63, 0x0a, 0xe1, 0x64, 0xd4, 0xc7, 0xa6, 0xce, 0xd4, 0x2f,
-	0x78, 0x05, 0xe4, 0xa9, 0x37, 0x20, 0x4d, 0x6c, 0x09, 0x58, 0xc7, 0x2e, 0xc5, 0xb6, 0xb5, 0xdd,
-	0xb8, 0x99, 0x95, 0x52, 0xee, 0xd1, 0xb0, 0xe1, 0x08, 0x6c, 0x8e, 0x13, 0x24, 0x0d, 0x11, 0x63,
-	0xc4, 0x39, 0x1d, 0x30, 0x4c, 0x4b, 0xf1, 0x6d, 0x6d, 0x37, 0x53, 0xbf, 0x6d, 0xbc, 0xb9, 0xc8,
-	0x8c, 0xfb, 0x3e, 0x08, 0xc7, 0x3d, 0x18, 0x43, 0x1c, 0xad, 0x98, 0x1b, 0xee, 0x6c, 0x15, 0xa4,
-	0x60, 0x43, 0xe5, 0x31, 0x12, 0x38, 0x21, 0x02, 0xdf, 0x5c, 0x26, 0xf0, 0x91, 0x84, 0x88, 0x84,
-	0x5d, 0xef, 0xcc, 0x52, 0xc0, 0xef, 0x35, 0xb0, 0x43, 0x47, 0x6e, 0xd3, 0xa2, 0x1d, 0x44, 0x6c,
-	0x8b, 0x32, 0xc4, 0x06, 0x34, 0x12, 0x7f, 0x55, 0xc4, 0x3f, 0x58, 0x26, 0xfe, 0xc3, 0x91, 0xdb,
-	0x7c, 0xc8, 0xb1, 0x1e, 0x0a, 0xa8, 0xc8, 0x3e, 0xb6, 0xe8, 0x22, 0x03, 0xf8, 0x8d, 0x06, 0x84,
-	0x85, 0x85, 0x9a, 0xcc, 0x19, 0x3a, 0x2c, 0x9a, 0x8b, 0xa4, 0xd8, 0xcb, 0x47, 0xcb, 0xee, 0xe5,
-	0x40, 0xe1, 0x44, 0x36, 0x52, 0xa6, 0x73, 0xb5, 0xf0, 0x3b, 0x0d, 0x6c, 0xfb, 0x77, 0xd1, 0xc3,
-	0x0c, 0xd9, 0x88, 0xa1, 0xc8, 0x46, 0x52, 0xcb, 0x27, 0x45, 0x5d, 0xca, 0x3d, 0x05, 0x15, 0x4d,
-	0x4a, 0x67, 0x91, 0x01, 0xfc, 0x0a, 0x94, 0x43, 0x95, 0x31, 0xac, 0x07, 0xf7, 0xa1, 0x2f, 0x5f,
-	0x95, 0x81, 0xe2, 0x78, 0x54, 0x0f, 0x57, 0x65, 0x67, 0xb6, 0xea, 0x30, 0x0b, 0xc0, 0x24, 0x56,
-	0xf5, 0xa9, 0x06, 0x8a, 0xc1, 0x36, 0xf3, 0xce, 0xb0, 0x0b, 0x37, 0x81, 0x2e, 0xab, 0xc7, 0xb1,
-	0x45, 0xa3, 0xae, 0x9a, 0x29, 0xb1, 0x6e, 0xd8, 0xf0, 0x26, 0xd8, 0xec, 0x22, 0xca, 0x2c, 0x82,
-	0x19, 0x71, 0xf0, 0x10, 0xdb, 0x96, 0x6a, 0xfc, 0x49, 0xff, 0xfd, 0x97, 0x1b, 0x98, 0xbe, 0xfe,
-	0x9e, 0x54, 0x07, 0x5c, 0xfb, 0xc4, 0x6b, 0x62, 0x4a, 0xc3, 0xae, 0xf1, 0x89, 0xeb, 0x03, 0x5f,
-	0x3f, 0x76, 0xad, 0x9e, 0x80, 0xc2, 0x54, 0x19, 0xc2, 0x03, 0x90, 0xf1, 0x6b, 0xdb, 0xe9, 0xc9,
-	0x79, 0x92, 0xa9, 0x97, 0x0d, 0x49, 0x05, 0x86, 0x4f, 0x05, 0xc6, 0x89, 0x4f, 0x05, 0x87, 0x89,
-	0x27, 0xbf, 0x5d, 0xd6, 0x4c, 0x20, 0x9d, 0xb8, 0xb8, 0xfa, 0x73, 0x0c, 0xac, 0x05, 0xce, 0xae,
-	0xc2, 0x51, 0xf8, 0x05, 0xb8, 0x14, 0x48, 0xb3, 0xb8, 0x21, 0x5a, 0xd2, 0xb6, 0xe3, 0xbb, 0x99,
-	0xfa, 0xfe, 0x32, 0x97, 0x32, 0x35, 0xb6, 0xcc, 0x22, 0x09, 0x0b, 0xe8, 0x3f, 0xc9, 0xe2, 0x26,
-	0xd0, 0x3b, 0x88, 0x5a, 0x3d, 0x8f, 0x60, 0x91, 0x34, 0xdd, 0x4c, 0x75, 0x10, 0xbd, 0xe7, 0x11,
-	0x0c, 0x2d, 0x70, 0x29, 0xd2, 0xf9, 0x6a, 0xd2, 0xec, 0xbf, 0x45, 0xa7, 0x9b, 0x85, 0xa9, 0xce,
-	0xae, 0xfe, 0x12, 0x4e, 0x98, 0x98, 0xb0, 0x6e, 0xcb, 0x83, 0x3b, 0x20, 0x3b, 0x99, 0xb1, 0xaa,
-	0x66, 0xd2, 0x66, 0x66, 0x2c, 0x6b, 0xd8, 0xf0, 0x32, 0xc8, 0x9c, 0x7b, 0xe4, 0xac, 0xd5, 0xf5,
-	0xce, 0xfd, 0x33, 0xa6, 0x4d, 0xe0, 0x8b, 0x1a, 0x36, 0x5c, 0x07, 0x49, 0x32, 0x70, 0xfd, 0x52,
-	0x48, 0x9b, 0xab, 0x64, 0xe0, 0x36, 0x6c, 0x78, 0x27, 0x48, 0x1a, 0x09, 0x41, 0x1a, 0xff, 0x5b,
-	0x4c, 0x1a, 0x33, 0x98, 0x62, 0x03, 0xa4, 0x7c, 0x8a, 0x58, 0x15, 0xc9, 0x4d, 0x32, 0x49, 0x0e,
-	0x25, 0x90, 0x1a, 0x62, 0x42, 0x1d, 0xcf, 0x15, 0x53, 0x28, 0x6e, 0xfa, 0x4b, 0x4e, 0x2e, 0x2d,
-	0x87, 0x50, 0x66, 0xe1, 0x21, 0x76, 0x19, 0xf7, 0x4c, 0x49, 0x72, 0x11, 0xd2, 0x4f, 0xb8, 0xb0,
-	0x61, 0xc3, 0x2a, 0xc8, 0xb9, 0xf8, 0xcb, 0x80, 0x91, 0x2e, 0x8c, 0x32, 0x5c, 0xe8, 0xdb, 0xec,
-	0x80, 0x2c, 0x6d, 0x76, 0xb0, 0x3d, 0xe8, 0x62, 0xd1, 0x50, 0x69, 0x69, 0x32, 0x96, 0x35, 0xec,
-	0xea, 0xb3, 0x38, 0xd8, 0x98, 0xc3, 0x2f, 0x10, 0x81, 0xb5, 0x49, 0x6e, 0xbd, 0x3e, 0x26, 0x22,
-	0xf5, 0x8a, 0x3f, 0x6f, 0x2c, 0x4e, 0xc5, 0x18, 0xf3, 0x33, 0xdf, 0xcf, 0x84, 0x6e, 0x44, 0x06,
-	0xf3, 0x20, 0x36, 0xbe, 0x92, 0x98, 0x63, 0xc3, 0x0f, 0x41, 0xc2, 0x71, 0x5b, 0x9e, 0x62, 0xc7,
-	0xdd, 0x49, 0x0c, 0x0e, 0x3e, 0xf6, 0x0f, 0x05, 0xe0, 0x65, 0x60, 0x0a, 0x2f, 0x78, 0x08, 0x92,
-	0x4d, 0xcf, 0x6d, 0x39, 0x6d, 0x55, 0x7a, 0xff, 0x5f, 0xc6, 0xff, 0x8e, 0xf0, 0x30, 0x95, 0x27,
-	0x6c, 0x01, 0x18, 0xec, 0x40, 0x85, 0x27, 0x49, 0xeb, 0xfd, 0x30, 0xde, 0x3c, 0x9a, 0x0e, 0xd4,
-	0xa9, 0x02, 0x0f, 0x36, 0xb5, 0x14, 0xc1, 0xab, 0x20, 0x2f, 0xb1, 0xad, 0x70, 0x19, 0xe4, 0xa4,
-	0xf4, 0x91, 0x2a, 0x86, 0xeb, 0xa0, 0xc8, 0x5f, 0x3a, 0xde, 0x10, 0x93, 0xb1, 0xa1, 0x2c, 0x87,
-	0x82, 0x2f, 0x57, 0xa6, 0xd5, 0xa7, 0x71, 0xb0, 0x3e, 0x93, 0xb1, 0xe1, 0x35, 0x50, 0x60, 0x88,
-	0xb4, 0x31, 0xb3, 0x9a, 0xdd, 0x01, 0x65, 0x98, 0xc8, 0x99, 0x92, 0x36, 0xf3, 0x52, 0x7c, 0x47,
-	0x49, 0x23, 0xdd, 0x14, 0x7b, 0x63, 0x37, 0xc5, 0x17, 0x74, 0x53, 0x22, 0xd8, 0x4d, 0xd1, 0xaa,
-	0x5e, 0x5d, 0xa6, 0xaa, 0x93, 0xd1, 0xaa, 0x0e, 0x74, 0x4e, 0x2a, 0xdc, 0x39, 0xb7, 0x40, 0x4a,
-	0x51, 0x8f, 0x28, 0xf5, 0x4c, 0x7d, 0x3b, 0x7c, 0x61, 0x4a, 0x19, 0x60, 0x2f, 0xd3, 0x77, 0x80,
-	0x47, 0xa0, 0xe0, 0xe2, 0x73, 0x8b, 0x6f, 0xdd, 0xc7, 0x00, 0x4b, 0x62, 0xe4, 0x5c, 0x7c, 0x6e,
-	0x0e, 0x5c, 0xb5, 0x3c, 0x4e, 0xe8, 0x7a, 0x31, 0x7d, 0x9c, 0xd0, 0x33, 0xc5, 0xec, 0x71, 0x42,
-	0xcf, 0x16, 0x73, 0xc7, 0x09, 0x3d, 0x57, 0xcc, 0x1f, 0x27, 0xf4, 0x7c, 0xb1, 0x50, 0xfd, 0x36,
-	0x06, 0xb6, 0x16, 0x52, 0xf8, 0xbf, 0xe5, 0xb6, 0xaa, 0x3f, 0x68, 0x60, 0x6b, 0xe1, 0x0b, 0x8f,
-	0xf7, 0x88, 0x7a, 0x66, 0xab, 0x4c, 0xa8, 0xf1, 0x9e, 0x93, 0x52, 0x95, 0x88, 0xd0, 0x9b, 0x21,
-	0x16, 0x7e, 0x33, 0x4c, 0x51, 0x75, 0xfc, 0x2d, 0xa8, 0xfa, 0xd7, 0x55, 0x50, 0x9e, 0xff, 0xf8,
-	0x7b, 0x97, 0x04, 0x14, 0x48, 0x5d, 0x22, 0x5c, 0xe8, 0xd3, 0x83, 0x7d, 0x35, 0x32, 0xd8, 0xe1,
-	0xa7, 0x20, 0x3f, 0x31, 0x11, 0x87, 0x4f, 0x2e, 0x79, 0xf8, 0xdc, 0xd8, 0x8f, 0x6b, 0xe0, 0x16,
-	0xe0, 0xd9, 0x20, 0x4c, 0x46, 0x92, 0x77, 0x98, 0x56, 0x12, 0xc1, 0x92, 0x59, 0x5f, 0x2d, 0xa2,
-	0xe8, 0x4b, 0x46, 0xc9, 0x28, 0x2f, 0x11, 0xe3, 0x01, 0x58, 0x13, 0x8f, 0x92, 0x0e, 0x46, 0x84,
-	0x9d, 0x62, 0xc4, 0x24, 0x56, 0x7a, 0x49, 0xac, 0x4b, 0xdc, 0xf9, 0xc8, 0xf7, 0x15, 0x88, 0xb7,
-	0x40, 0xca, 0xc6, 0x0c, 0x39, 0x5d, 0x3a, 0xbb, 0x8d, 0xe5, 0xf7, 0x2d, 0xef, 0xe2, 0x07, 0x68,
-	0xd4, 0xf5, 0x90, 0x4d, 0x4d, 0xdf, 0x81, 0xe7, 0x1d, 0x31, 0x6e, 0xcd, 0x4a, 0x19, 0x59, 0x4e,
-	0x6a, 0xc9, 0x0f, 0x2b, 0xf6, 0xa9, 0x3e, 0x3e, 0x4b, 0xd9, 0x59, 0xd0, 0x4a, 0xc9, 0xb1, 0xef,
-	0xca, 0x9f, 0x66, 0x86, 0x7b, 0xa9, 0x05, 0xbc, 0x01, 0xfe, 0x23, 0x40, 0x78, 0x01, 0x60, 0x62,
-	0x39, 0x36, 0x76, 0x99, 0xc3, 0x46, 0xa5, 0x9c, 0xb8, 0x7b, 0xc8, 0x75, 0x8f, 0x85, 0xaa, 0xa1,
-	0x34, 0xf0, 0x31, 0x28, 0xa8, 0x9b, 0x1f, 0xcf, 0xa6, 0xbc, 0x88, 0x6c, 0xcc, 0x24, 0xe1, 0xc0,
-	0x88, 0x52, 0xdc, 0xe0, 0x4f, 0xaa, 0xfc, 0x30, 0xb4, 0xae, 0xfe, 0x11, 0x03, 0x1b, 0x73, 0xde,
-	0xf1, 0xef, 0x72, 0xba, 0xb4, 0xc0, 0xfa, 0xd4, 0x79, 0x2c, 0x87, 0xe1, 0x1e, 0xff, 0x36, 0xe4,
-	0x2f, 0xdd, 0xfa, 0xdf, 0x3b, 0x55, 0x83, 0xe1, 0x9e, 0xb9, 0x36, 0x8c, 0xc8, 0x28, 0xfc, 0x00,
-	0x24, 0xc5, 0x68, 0xf2, 0x3f, 0xf4, 0xe6, 0xd6, 0xc0, 0xc7, 0x88, 0xa1, 0xc3, 0xae, 0x77, 0x6a,
-	0x2a, 0x7b, 0x78, 0x17, 0xe4, 0x7d, 0x36, 0x50, 0x08, 0xa9, 0x25, 0x11, 0xb2, 0x92, 0x0c, 0xc4,
-	0xf8, 0xa3, 0xc7, 0x09, 0x5d, 0x2b, 0xc6, 0x0e, 0x9d, 0xe7, 0x2f, 0x2b, 0x2b, 0x2f, 0x5e, 0x56,
-	0x56, 0x5e, 0xbf, 0xac, 0x68, 0x5f, 0x5f, 0x54, 0xb4, 0x9f, 0x2e, 0x2a, 0xda, 0xb3, 0x8b, 0x8a,
-	0xf6, 0xfc, 0xa2, 0xa2, 0xfd, 0x7e, 0x51, 0xd1, 0xfe, 0xbc, 0xa8, 0xac, 0xbc, 0xbe, 0xa8, 0x68,
-	0x4f, 0x5e, 0x55, 0x56, 0x9e, 0xbf, 0xaa, 0xac, 0xbc, 0x78, 0x55, 0x59, 0xf9, 0x7c, 0xbf, 0xed,
-	0x4d, 0xa2, 0x39, 0xde, 0xfc, 0xff, 0xad, 0x6e, 0x13, 0xdc, 0x57, 0xab, 0xd3, 0xa4, 0x68, 0x92,
-	0xfd, 0xbf, 0x02, 0x00, 0x00, 0xff, 0xff, 0x9f, 0xd0, 0x18, 0x88, 0xef, 0x12, 0x00, 0x00,
+	// 1537 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x58, 0x4f, 0x6f, 0x1b, 0x45,
+	0x14, 0xcf, 0xda, 0x8e, 0xff, 0x3c, 0xff, 0xed, 0x84, 0x10, 0xc7, 0x28, 0x6e, 0x62, 0xb5, 0x34,
+	0x45, 0xc8, 0x6e, 0x92, 0x03, 0xb4, 0x45, 0x48, 0x49, 0xa1, 0xc4, 0x91, 0x5a, 0xaa, 0x6d, 0xd4,
+	0x4a, 0x5c, 0x96, 0x89, 0x77, 0x6c, 0xaf, 0x62, 0xef, 0x5a, 0x33, 0x63, 0x07, 0x73, 0x42, 0xe2,
+	0x80, 0x84, 0x40, 0xea, 0x77, 0x80, 0x03, 0x27, 0x3e, 0x47, 0x8f, 0xbd, 0x20, 0x95, 0x13, 0x34,
+	0xe5, 0xc0, 0xb1, 0x7c, 0x03, 0x34, 0xb3, 0xb3, 0xf6, 0xae, 0xd7, 0x76, 0xdd, 0xa2, 0x9e, 0xb8,
+	0x79, 0xde, 0x7b, 0xbf, 0xdf, 0x9b, 0x79, 0xf3, 0xfe, 0xcc, 0x1a, 0xae, 0x71, 0xd2, 0xed, 0x39,
+	0x14, 0x77, 0x6a, 0x8c, 0xd0, 0x01, 0xa1, 0x35, 0xdc, 0xb3, 0x6a, 0x94, 0xf4, 0x3a, 0x56, 0x03,
+	0x73, 0xcb, 0xb1, 0x6b, 0x83, 0x9d, 0x5a, 0x97, 0x30, 0x86, 0x5b, 0xa4, 0xda, 0xa3, 0x0e, 0x77,
+	0x50, 0xc5, 0x43, 0x54, 0x5d, 0x44, 0x15, 0xf7, 0xac, 0xaa, 0x0f, 0x51, 0x1d, 0xec, 0x94, 0x2e,
+	0xb6, 0x1c, 0xa7, 0xd5, 0x21, 0x35, 0x89, 0x38, 0xe9, 0x37, 0x6b, 0xdc, 0xea, 0x12, 0xc6, 0x71,
+	0xb7, 0xe7, 0x92, 0x94, 0xb6, 0x4c, 0xd2, 0x23, 0xb6, 0x49, 0xec, 0x86, 0x45, 0x58, 0xad, 0xe5,
+	0xb4, 0x1c, 0x29, 0x97, 0xbf, 0x94, 0x49, 0x75, 0xda, 0xce, 0x88, 0xdd, 0xef, 0x32, 0xb1, 0x27,
+	0xbf, 0x43, 0xd7, 0xfe, 0xca, 0x5c, 0x7b, 0x8e, 0xd9, 0xa9, 0x32, 0x7c, 0x7f, 0x9a, 0x61, 0xdb,
+	0x62, 0xdc, 0xa1, 0xc3, 0xd0, 0x71, 0x4b, 0x97, 0x46, 0xd6, 0xc2, 0xac, 0xe1, 0x74, 0xbb, 0x53,
+	0x82, 0xe2, 0x73, 0x2e, 0xac, 0x6c, 0xdc, 0x25, 0xac, 0x87, 0x1b, 0x24, 0x6c, 0x78, 0x35, 0x60,
+	0x38, 0x2f, 0xd0, 0xa5, 0xcb, 0x01, 0xd3, 0x99, 0x1b, 0x0c, 0x9a, 0x35, 0xb1, 0xd5, 0xe9, 0xd3,
+	0xb0, 0xe3, 0xca, 0x3f, 0x09, 0xc8, 0xeb, 0x63, 0x77, 0xc7, 0x98, 0x9d, 0xa2, 0xbb, 0x90, 0x12,
+	0x71, 0x31, 0xf8, 0xb0, 0x47, 0x8a, 0xda, 0xa6, 0xb6, 0x9d, 0xdb, 0xdd, 0xa9, 0x4e, 0xbb, 0x5e,
+	0x19, 0xc6, 0xea, 0x60, 0xa7, 0x3a, 0xc1, 0x70, 0x3c, 0xec, 0x11, 0x3d, 0xc9, 0xd5, 0x2f, 0x74,
+	0x09, 0x72, 0xcc, 0xe9, 0xd3, 0x06, 0x31, 0x24, 0xad, 0x65, 0x16, 0x23, 0x9b, 0xda, 0x76, 0x54,
+	0xcf, 0xb8, 0x52, 0x81, 0xa8, 0x9b, 0x68, 0x08, 0xeb, 0xa3, 0x00, 0xb9, 0x86, 0x98, 0x73, 0x6a,
+	0x9d, 0xf4, 0x39, 0x61, 0xc5, 0xe8, 0xa6, 0xb6, 0x9d, 0xde, 0xbd, 0x59, 0x7d, 0x79, 0x92, 0x55,
+	0xef, 0x7a, 0x24, 0x82, 0x77, 0x7f, 0x44, 0x71, 0xb8, 0xa4, 0xaf, 0xd9, 0xd3, 0x55, 0x88, 0xc1,
+	0x9a, 0x8a, 0x63, 0xc8, 0x71, 0x4c, 0x3a, 0xbe, 0xbe, 0x88, 0xe3, 0x43, 0x97, 0x22, 0xe4, 0x76,
+	0xb5, 0x3d, 0x4d, 0x81, 0x7e, 0xd4, 0x60, 0x8b, 0x0d, 0xed, 0x86, 0xc1, 0xda, 0x98, 0x9a, 0x06,
+	0xe3, 0x98, 0xf7, 0x59, 0xc8, 0xff, 0xb2, 0xf4, 0xbf, 0xbf, 0x88, 0xff, 0xfb, 0x43, 0xbb, 0x71,
+	0x5f, 0x70, 0xdd, 0x97, 0x54, 0xa1, 0x7d, 0x6c, 0xb0, 0x79, 0x06, 0xe8, 0x5b, 0x0d, 0xa4, 0x85,
+	0x81, 0x1b, 0xdc, 0x1a, 0x58, 0x3c, 0x1c, 0x8b, 0xb8, 0xdc, 0xcb, 0xc7, 0x8b, 0xee, 0x65, 0x5f,
+	0xf1, 0x84, 0x36, 0x52, 0x62, 0x33, 0xb5, 0xe8, 0x07, 0x0d, 0x36, 0xbd, 0xbb, 0xe8, 0x12, 0x8e,
+	0x4d, 0xcc, 0x71, 0x68, 0x23, 0x89, 0xc5, 0x83, 0xa2, 0x2e, 0xe5, 0x8e, 0xa2, 0x0a, 0x07, 0xa5,
+	0x3d, 0xcf, 0x00, 0x7d, 0x0d, 0xa5, 0x40, 0x66, 0x0c, 0x76, 0xfd, 0xfb, 0x48, 0x2e, 0x9e, 0x95,
+	0xbe, 0xe4, 0x78, 0xb0, 0x1b, 0xcc, 0xca, 0xf6, 0x74, 0x15, 0xaa, 0x43, 0x7e, 0x60, 0x31, 0xeb,
+	0xc4, 0xea, 0xc8, 0xcb, 0xb0, 0xba, 0xa4, 0x98, 0x92, 0x0e, 0x4b, 0x55, 0xb7, 0x8f, 0x56, 0xbd,
+	0x3e, 0x5a, 0x3d, 0xf6, 0xfa, 0xe8, 0x41, 0xec, 0xd1, 0x1f, 0x17, 0x35, 0x3d, 0x37, 0x06, 0x0a,
+	0xd5, 0x41, 0x06, 0x60, 0xbc, 0xed, 0xca, 0xf7, 0x11, 0x28, 0xf8, 0x2b, 0xd6, 0x39, 0x25, 0x36,
+	0x5a, 0x87, 0xa4, 0x9b, 0x88, 0x96, 0x29, 0x6b, 0x7e, 0x59, 0x4f, 0xc8, 0x75, 0xdd, 0x44, 0xd7,
+	0x61, 0xbd, 0x83, 0x19, 0x37, 0x28, 0xe1, 0xd4, 0x22, 0x03, 0x62, 0x1a, 0xaa, 0x87, 0x8c, 0x4b,
+	0xf9, 0x6d, 0x61, 0xa0, 0x7b, 0xfa, 0x3b, 0xae, 0xda, 0x07, 0xed, 0x51, 0xa7, 0x41, 0x18, 0x0b,
+	0x42, 0xa3, 0x63, 0xe8, 0x3d, 0x4f, 0x3f, 0x86, 0x12, 0x28, 0x4f, 0x40, 0x27, 0xa3, 0x11, 0x5b,
+	0x30, 0x1a, 0xef, 0x04, 0x3c, 0x3c, 0x08, 0x84, 0xa6, 0x72, 0x0c, 0xf9, 0x89, 0xc2, 0x41, 0xfb,
+	0x90, 0xf6, 0xaa, 0x51, 0xb8, 0xd1, 0x16, 0x74, 0x03, 0x2e, 0x48, 0xb2, 0xfe, 0x1a, 0x81, 0x15,
+	0x5f, 0x88, 0xd5, 0xa9, 0x18, 0xfa, 0x12, 0x2e, 0xf8, 0x12, 0x43, 0xe6, 0x14, 0x2b, 0x6a, 0x9b,
+	0xd1, 0xed, 0xf4, 0xee, 0xde, 0x22, 0x69, 0x34, 0xd1, 0x68, 0xf5, 0x02, 0x0d, 0x0a, 0xd8, 0x7f,
+	0xb9, 0xac, 0x75, 0x48, 0xb6, 0x31, 0x33, 0xba, 0x0e, 0x25, 0xf2, 0x6e, 0x92, 0x7a, 0xa2, 0x8d,
+	0xd9, 0x1d, 0x87, 0x12, 0x64, 0xc0, 0x85, 0x50, 0xaf, 0x52, 0xf1, 0xdf, 0x7b, 0x8d, 0xde, 0xa4,
+	0xe7, 0x27, 0x7a, 0x51, 0xe5, 0xb7, 0x60, 0xc0, 0xe4, 0x4c, 0xb0, 0x9b, 0x0e, 0xda, 0x82, 0xcc,
+	0x78, 0x2a, 0xa8, 0xd4, 0x4c, 0xe9, 0xe9, 0x91, 0xac, 0x6e, 0xa2, 0x8b, 0x90, 0x3e, 0x73, 0xe8,
+	0x69, 0xb3, 0xe3, 0x9c, 0x79, 0x67, 0x4c, 0xe9, 0xe0, 0x89, 0xea, 0x26, 0x5a, 0x85, 0x38, 0xed,
+	0xdb, 0x5e, 0xc6, 0xa5, 0xf4, 0x65, 0xda, 0xb7, 0xeb, 0x26, 0xba, 0xe5, 0x1f, 0x73, 0x31, 0x39,
+	0xe6, 0xde, 0x9d, 0x3f, 0xe6, 0xa6, 0xcc, 0xb6, 0x35, 0x48, 0x78, 0x43, 0x6d, 0x59, 0x06, 0x37,
+	0xce, 0xdd, 0x71, 0x56, 0x84, 0xc4, 0x80, 0x50, 0x66, 0x39, 0xb6, 0xec, 0x9b, 0x51, 0xdd, 0x5b,
+	0x8a, 0x71, 0xd8, 0xb4, 0x28, 0xe3, 0x06, 0x19, 0x10, 0x9b, 0x0b, 0x64, 0xc2, 0x1d, 0x87, 0x52,
+	0xfa, 0xa9, 0x10, 0xd6, 0x4d, 0x54, 0x81, 0xac, 0x4d, 0xbe, 0xf2, 0x19, 0x25, 0xa5, 0x51, 0x5a,
+	0x08, 0x3d, 0x9b, 0x2d, 0xc8, 0xb0, 0x46, 0x9b, 0x98, 0xfd, 0x0e, 0x91, 0x75, 0x9b, 0x72, 0x4d,
+	0x46, 0xb2, 0xba, 0x59, 0x79, 0x1c, 0x85, 0xb5, 0x19, 0x13, 0x11, 0x61, 0x58, 0x19, 0xc7, 0xd6,
+	0xe9, 0x11, 0x2a, 0x43, 0xaf, 0x26, 0xfe, 0xb5, 0xf9, 0xa1, 0x18, 0x71, 0x7e, 0xee, 0xe1, 0x74,
+	0x64, 0x87, 0x64, 0x28, 0x07, 0x91, 0xd1, 0x95, 0x44, 0x2c, 0x13, 0x7d, 0x04, 0x31, 0xcb, 0x6e,
+	0x3a, 0x6a, 0x9e, 0x6f, 0x8f, 0x7d, 0x08, 0xf2, 0x11, 0x3e, 0xe0, 0x40, 0xa4, 0x81, 0x2e, 0x51,
+	0xe8, 0x00, 0xe2, 0x0d, 0xc7, 0x6e, 0x5a, 0x2d, 0x95, 0x7a, 0xef, 0x2d, 0x82, 0xbf, 0x25, 0x11,
+	0xba, 0x42, 0xa2, 0x26, 0x20, 0x7f, 0x05, 0x2a, 0x3e, 0x77, 0xcc, 0x7e, 0x10, 0xe4, 0x9b, 0xf5,
+	0xb0, 0xf0, 0xe5, 0xa9, 0x22, 0xf7, 0x17, 0xb5, 0x2b, 0x42, 0x97, 0x21, 0xe7, 0x72, 0x1b, 0xc1,
+	0x34, 0xc8, 0xba, 0xd2, 0x07, 0x2a, 0x19, 0xae, 0x42, 0x41, 0xbc, 0xcd, 0x9c, 0x01, 0xa1, 0x23,
+	0x43, 0x37, 0x1d, 0xf2, 0x9e, 0x5c, 0x99, 0x56, 0x7e, 0x8a, 0xc2, 0xea, 0xd4, 0x37, 0x06, 0xba,
+	0x02, 0x79, 0x8e, 0x69, 0x8b, 0x70, 0xa3, 0xd1, 0xe9, 0x33, 0x4e, 0xa8, 0xdb, 0x53, 0x52, 0x7a,
+	0xce, 0x15, 0xdf, 0x52, 0xd2, 0x50, 0x35, 0x45, 0x5e, 0x5a, 0x4d, 0xd1, 0x39, 0xd5, 0x14, 0xf3,
+	0x57, 0x53, 0x38, 0xab, 0x97, 0x17, 0xc9, 0xea, 0x78, 0x38, 0xab, 0x7d, 0x95, 0x93, 0x08, 0x56,
+	0xce, 0x0d, 0x48, 0xa8, 0x61, 0xa9, 0x26, 0xe1, 0x66, 0xf0, 0xc2, 0x94, 0xd2, 0x37, 0x6f, 0x75,
+	0x0f, 0x80, 0x0e, 0x21, 0x6f, 0x93, 0x33, 0x43, 0x6c, 0xdd, 0xe3, 0x80, 0x05, 0x39, 0xb2, 0x36,
+	0x39, 0xd3, 0xfb, 0xb6, 0x5a, 0x1e, 0xc5, 0x92, 0xc9, 0x42, 0xea, 0x28, 0x96, 0x4c, 0x17, 0x32,
+	0x47, 0xb1, 0x64, 0xa6, 0x90, 0x3d, 0x8a, 0x25, 0xb3, 0x85, 0xdc, 0x51, 0x2c, 0x99, 0x2b, 0xe4,
+	0x2b, 0xdf, 0x45, 0x60, 0x63, 0xee, 0xa3, 0xe3, 0xff, 0x72, 0x5b, 0x95, 0x9f, 0x35, 0xd8, 0x98,
+	0xfb, 0x26, 0x15, 0x35, 0xa2, 0x3e, 0x0c, 0x54, 0x24, 0x54, 0x7b, 0xcf, 0xba, 0x52, 0x15, 0x88,
+	0xc0, 0xd3, 0x24, 0x12, 0x7c, 0x9a, 0x4c, 0x8c, 0xea, 0xe8, 0x6b, 0x8c, 0xea, 0xdf, 0x97, 0xa1,
+	0x34, 0xfb, 0xb9, 0xfa, 0x26, 0x07, 0x90, 0x2f, 0x74, 0xb1, 0x60, 0xa2, 0x4f, 0x36, 0xf6, 0xe5,
+	0x50, 0x63, 0x47, 0x9f, 0x41, 0x6e, 0x6c, 0x22, 0x0f, 0x1f, 0x5f, 0xf0, 0xf0, 0xd9, 0x11, 0x4e,
+	0x68, 0xd0, 0x06, 0x88, 0x68, 0x50, 0xee, 0x7a, 0x72, 0xef, 0x30, 0xa5, 0x24, 0x72, 0x4a, 0x66,
+	0x3c, 0xb5, 0xf4, 0x92, 0x5c, 0xd0, 0x4b, 0x5a, 0xa1, 0xa4, 0x8f, 0x7b, 0xb0, 0x22, 0x1f, 0x25,
+	0x6d, 0x82, 0x29, 0x3f, 0x21, 0x98, 0xbf, 0xda, 0x73, 0xf6, 0x82, 0x00, 0x1f, 0x7a, 0x58, 0xc9,
+	0x78, 0x03, 0x12, 0x26, 0xe1, 0xd8, 0xea, 0xb0, 0xe9, 0x65, 0xec, 0x7e, 0x91, 0x8b, 0x2a, 0xbe,
+	0x87, 0x87, 0x1d, 0x07, 0x9b, 0x4c, 0xf7, 0x00, 0x22, 0xee, 0x98, 0x0b, 0x6b, 0x5e, 0x4c, 0xbb,
+	0xe9, 0xa4, 0x96, 0xe2, 0xb0, 0x72, 0x9f, 0xea, 0x73, 0xb9, 0x98, 0x99, 0x46, 0xad, 0x94, 0x82,
+	0xfb, 0xb6, 0xfb, 0x53, 0x4f, 0x0b, 0x94, 0x5a, 0xa0, 0x6b, 0xf0, 0x96, 0x24, 0x11, 0x09, 0x40,
+	0xa8, 0x61, 0x99, 0xc4, 0xe6, 0x16, 0x1f, 0x16, 0xb3, 0xf2, 0xee, 0x91, 0xd0, 0x3d, 0x94, 0xaa,
+	0xba, 0xd2, 0xa0, 0x87, 0x90, 0x57, 0x37, 0x3f, 0xea, 0x4d, 0x39, 0xe9, 0xb9, 0x3a, 0x75, 0x08,
+	0xfb, 0x5a, 0x94, 0x9a, 0x0d, 0x5e, 0xa7, 0xca, 0x0d, 0x02, 0xeb, 0xca, 0x5f, 0x11, 0x58, 0x9b,
+	0xf1, 0xe5, 0xf1, 0x26, 0xbb, 0x4b, 0x13, 0x56, 0x27, 0xce, 0x63, 0x58, 0x9c, 0x74, 0xc5, 0xd7,
+	0xac, 0x78, 0xe9, 0xee, 0xbe, 0xda, 0xa9, 0xea, 0x9c, 0x74, 0xf5, 0x95, 0x41, 0x48, 0xc6, 0xd0,
+	0x87, 0x10, 0x97, 0xad, 0xc9, 0xfb, 0x34, 0x9d, 0x99, 0x03, 0x9f, 0x60, 0x8e, 0x0f, 0x3a, 0xce,
+	0x89, 0xae, 0xec, 0xd1, 0x6d, 0xc8, 0x79, 0xd3, 0x40, 0x31, 0x24, 0x16, 0x64, 0xc8, 0xb8, 0xc3,
+	0x40, 0xb6, 0x3f, 0x76, 0x14, 0x4b, 0x6a, 0x85, 0xc8, 0x81, 0xf5, 0xe4, 0x59, 0x79, 0xe9, 0xe9,
+	0xb3, 0xf2, 0xd2, 0x8b, 0x67, 0x65, 0xed, 0x9b, 0xf3, 0xb2, 0xf6, 0xcb, 0x79, 0x59, 0x7b, 0x7c,
+	0x5e, 0xd6, 0x9e, 0x9c, 0x97, 0xb5, 0x3f, 0xcf, 0xcb, 0xda, 0xdf, 0xe7, 0xe5, 0xa5, 0x17, 0xe7,
+	0x65, 0xed, 0xd1, 0xf3, 0xf2, 0xd2, 0x93, 0xe7, 0xe5, 0xa5, 0xa7, 0xcf, 0xcb, 0x4b, 0x5f, 0xec,
+	0xb5, 0x9c, 0xb1, 0x37, 0xcb, 0x99, 0xfd, 0x4f, 0xdb, 0x4d, 0x4a, 0x7a, 0x6a, 0x75, 0x12, 0x97,
+	0x45, 0xb2, 0xf7, 0x6f, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa0, 0xdf, 0x1d, 0x36, 0xa1, 0x13, 0x00,
+	0x00,
 }

 func (this *ReplicationTask) Equal(that interface{}) bool {
@@ -1224,6 +1244,13 @@ func (this *ReplicationTask) Equal(that interface{}) bool {
 	} else if !this.Attributes.Equal(that1.Attributes) {
 		return false
 	}
+	if that1.VisibilityTime == nil {
+		if this.VisibilityTime != nil {
+			return false
+		}
+	} else if !this.VisibilityTime.Equal(*that1.VisibilityTime) {
+		return false
+	}
 	return true
 }
 func (this *ReplicationTask_NamespaceTaskAttributes) Equal(that interface{}) bool {
@@ -1398,6 +1425,13 @@ func (this *ReplicationToken) Equal(that interface{}) bool {
 	if this.LastProcessedMessageId != that1.LastProcessedMessageId {
 		return false
 	}
+	if that1.LastProcessedVisibilityTime == nil {
+		if this.LastProcessedVisibilityTime != nil {
+			return false
+		}
+	} else if !this.LastProcessedVisibilityTime.Equal(*that1.LastProcessedVisibilityTime) {
+		return false
+	}
 	return true
 }
 func (this *SyncShardStatus) Equal(that interface{}) bool {
@@ -1813,13 +1847,14 @@ func (this *ReplicationTask) GoString() string {
 	if this == nil {
 		return "nil"
 	}
-	s := make([]string, 0, 12)
+	s := make([]string, 0, 13)
 	s = append(s, "&repication.ReplicationTask{")
 	s = append(s, "TaskType: "+fmt.Sprintf("%#v", this.TaskType)+",\n")
 	s = append(s, "SourceTaskId: "+fmt.Sprintf("%#v", this.SourceTaskId)+",\n")
 	if this.Attributes != nil {
 		s = append(s, "Attributes: "+fmt.Sprintf("%#v", this.Attributes)+",\n")
 	}
+	s = append(s, "VisibilityTime: "+fmt.Sprintf("%#v", this.VisibilityTime)+",\n")
 	s = append(s, "}")
 	return strings.Join(s, "")
 }
@@ -1875,11 +1910,12 @@ func (this *ReplicationToken) GoString() string {
 	if this == nil {
 		return "nil"
 	}
-	s := make([]string, 0, 7)
+	s := make([]string, 0, 8)
 	s = append(s, "&repication.ReplicationToken{")
 	s = append(s, "ShardId: "+fmt.Sprintf("%#v", this.ShardId)+",\n")
 	s = append(s, "LastRetrievedMessageId: "+fmt.Sprintf("%#v", this.LastRetrievedMessageId)+",\n")
 	s = append(s, "LastProcessedMessageId: "+fmt.Sprintf("%#v", this.LastProcessedMessageId)+",\n")
+	s = append(s, "LastProcessedVisibilityTime: "+fmt.Sprintf("%#v", this.LastProcessedVisibilityTime)+",\n")
 	s = append(s, "}")
 	return strings.Join(s, "")
 }
@@ -2078,6 +2114,16 @@ func (m *ReplicationTask) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.VisibilityTime != nil {
+		n1, err1 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.VisibilityTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.VisibilityTime):])
+		if err1 != nil {
+			return 0, err1
+		}
+		i -= n1
+		i = encodeVarintMessage(dAtA, i, uint64(n1))
+		i--
+		dAtA[i] = 0x4a
+	}
 	if m.Attributes != nil {
 		{
 			size := m.Attributes.Size()
@@ -2246,6 +2292,16 @@ func (m *ReplicationToken) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.LastProcessedVisibilityTime != nil {
+		n8, err8 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.LastProcessedVisibilityTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.LastProcessedVisibilityTime):])
+		if err8 != nil {
+			return 0, err8
+		}
+		i -= n8
+		i = encodeVarintMessage(dAtA, i, uint64(n8))
+		i--
+		dAtA[i] = 0x22
+	}
 	if m.LastProcessedMessageId != 0 {
 		i = encodeVarintMessage(dAtA, i, uint64(m.LastProcessedMessageId))
 		i--
@@ -2285,12 +2341,12 @@ func (m *SyncShardStatus) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	var l int
 	_ = l
 	if m.StatusTime != nil {
-		n7, err7 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.StatusTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.StatusTime):])
-		if err7 != nil {
-			return 0, err7
+		n9, err9 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.StatusTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.StatusTime):])
+		if err9 != nil {
+			return 0, err9
 		}
-		i -= n7
-		i = encodeVarintMessage(dAtA, i, uint64(n7))
+		i -= n9
+		i = encodeVarintMessage(dAtA, i, uint64(n9))
 		i--
 		dAtA[i] = 0xa
 	}
@@ -2697,12 +2753,12 @@ func (m *SyncShardStatusTaskAttributes) MarshalToSizedBuffer(dAtA []byte) (int,
 	var l int
 	_ = l
 	if m.StatusTime != nil {
-		n14, err14 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.StatusTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.StatusTime):])
-		if err14 != nil {
-			return 0, err14
+		n16, err16 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.StatusTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.StatusTime):])
+		if err16 != nil {
+			return 0, err16
 		}
-		i -= n14
-		i = encodeVarintMessage(dAtA, i, uint64(n14))
+		i -= n16
+		i = encodeVarintMessage(dAtA, i, uint64(n16))
 		i--
 		dAtA[i] = 0x1a
 	}
@@ -2790,22 +2846,22 @@ func (m *SyncActivityTaskAttributes) MarshalToSizedBuffer(dAtA []byte) (int, err
 		dAtA[i] = 0x52
 	}
 	if m.LastHeartbeatTime != nil {
-		n18, err18 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.LastHeartbeatTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.LastHeartbeatTime):])
-		if err18 != nil {
-			return 0, err18
+		n20, err20 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.LastHeartbeatTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.LastHeartbeatTime):])
+		if err20 != nil {
+			return 0, err20
 		}
-		i -= n18
-		i = encodeVarintMessage(dAtA, i, uint64(n18))
+		i -= n20
+		i = encodeVarintMessage(dAtA, i, uint64(n20))
 		i--
 		dAtA[i] = 0x4a
 	}
 	if m.StartedTime != nil {
-		n19, err19 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.StartedTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.StartedTime):])
-		if err19 != nil {
-			return 0, err19
+		n21, err21 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.StartedTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.StartedTime):])
+		if err21 != nil {
+			return 0, err21
 		}
-		i -= n19
-		i = encodeVarintMessage(dAtA, i, uint64(n19))
+		i -= n21
+		i = encodeVarintMessage(dAtA, i, uint64(n21))
 		i--
 		dAtA[i] = 0x42
 	}
@@ -2815,12 +2871,12 @@ func (m *SyncActivityTaskAttributes) MarshalToSizedBuffer(dAtA []byte) (int, err
 		dAtA[i] = 0x38
 	}
 	if m.ScheduledTime != nil {
-		n20, err20 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.ScheduledTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.ScheduledTime):])
-		if err20 != nil {
-			return 0, err20
+		n22, err22 := github_com_gogo_protobuf_types.StdTimeMarshalTo(*m.ScheduledTime, dAtA[i-github_com_gogo_protobuf_types.SizeOfStdTime(*m.ScheduledTime):])
+		if err22 != nil {
+			return 0, err22
 		}
-		i -= n20
-		i = encodeVarintMessage(dAtA, i, uint64(n20))
+		i -= n22
+		i = encodeVarintMessage(dAtA, i, uint64(n22))
 		i--
 		dAtA[i] = 0x32
 	}
@@ -2966,6 +3022,10 @@ func (m *ReplicationTask) Size() (n int) {
 	if m.Attributes != nil {
 		n += m.Attributes.Size()
 	}
+	if m.VisibilityTime != nil {
+		l = github_com_gogo_protobuf_types.SizeOfStdTime(*m.VisibilityTime)
+		n += 1 + l + sovMessage(uint64(l))
+	}
 	return n
 }

@@ -3056,6 +3116,10 @@ func (m *ReplicationToken) Size() (n int) {
 	if m.LastProcessedMessageId != 0 {
 		n += 1 + sovMessage(uint64(m.LastProcessedMessageId))
 	}
+	if m.LastProcessedVisibilityTime != nil {
+		l = github_com_gogo_protobuf_types.SizeOfStdTime(*m.LastProcessedVisibilityTime)
+		n += 1 + l + sovMessage(uint64(l))
+	}
 	return n
 }

@@ -3380,6 +3444,7 @@ func (this *ReplicationTask) String() string {
 		`TaskType:` + fmt.Sprintf("%v", this.TaskType) + `,`,
 		`SourceTaskId:` + fmt.Sprintf("%v", this.SourceTaskId) + `,`,
 		`Attributes:` + fmt.Sprintf("%v", this.Attributes) + `,`,
+		`VisibilityTime:` + strings.Replace(fmt.Sprintf("%v", this.VisibilityTime), "Timestamp", "types.Timestamp", 1) + `,`,
 		`}`,
 	}, "")
 	return s
@@ -3452,6 +3517,7 @@ func (this *ReplicationToken) String() string {
 		`ShardId:` + fmt.Sprintf("%v", this.ShardId) + `,`,
 		`LastRetrievedMessageId:` + fmt.Sprintf("%v", this.LastRetrievedMessageId) + `,`,
 		`LastProcessedMessageId:` + fmt.Sprintf("%v", this.LastProcessedMessageId) + `,`,
+		`LastProcessedVisibilityTime:` + strings.Replace(fmt.Sprintf("%v", this.LastProcessedVisibilityTime), "Timestamp", "types.Timestamp", 1) + `,`,
 		`}`,
 	}, "")
 	return s
@@ -3892,6 +3958,42 @@ func (m *ReplicationTask) Unmarshal(dAtA []byte) error {
 			}
 			m.Attributes = &ReplicationTask_HistoryTaskV2Attributes{v}
 			iNdEx = postIndex
+		case 9:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field VisibilityTime", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowMessage
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthMessage
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthMessage
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.VisibilityTime == nil {
+				m.VisibilityTime = new(time.Time)
+			}
+			if err := github_com_gogo_protobuf_types.StdTimeUnmarshal(m.VisibilityTime, dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipMessage(dAtA[iNdEx:])
@@ -4002,6 +4104,42 @@ func (m *ReplicationToken) Unmarshal(dAtA []byte) error {
 					break
 				}
 			}
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LastProcessedVisibilityTime", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowMessage
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthMessage
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthMessage
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.LastProcessedVisibilityTime == nil {
+				m.LastProcessedVisibilityTime = new(time.Time)
+			}
+			if err := github_com_gogo_protobuf_types.StdTimeUnmarshal(m.LastProcessedVisibilityTime, dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipMessage(dAtA[iNdEx:])
--- a/proto/internal/temporal/server/api/replication/v1/message.proto
+++ b/proto/internal/temporal/server/api/replication/v1/message.proto
@@ -51,6 +51,7 @@ message ReplicationTask {
        HistoryMetadataTaskAttributes history_metadata_task_attributes = 7;
        HistoryTaskV2Attributes history_task_v2_attributes = 8;
    }
+    google.protobuf.Timestamp visibility_time = 9 [(gogoproto.stdtime) = true];
 }

 message ReplicationToken {
@@ -60,6 +61,8 @@ message ReplicationToken {
    // lastProcessedMessageId is the last messageId that is processed on the passive side.
    // This can be different than lastRetrievedMessageId if passive side supports prefetching messages.
    int64 last_processed_message_id = 3;
+    // The VisibilityTime of last processed ReplicationTask
+    google.protobuf.Timestamp last_processed_visibility_time = 4 [(gogoproto.stdtime) = true];
 }

 message SyncShardStatus {
--- a/service/history/handler.go
+++ b/service/history/handler.go
@@ -1171,6 +1171,7 @@ func (h *Handler) GetReplicationMessages(ctx context.Context, request *historyse
 				ctx,
 				request.GetClusterName(),
 				token.GetLastProcessedMessageId(),
+				timestamp.TimeValue(token.LastProcessedVisibilityTime),
 				token.GetLastRetrievedMessageId(),
 			)
 			if err != nil {
--- a/service/history/historyEngine.go
+++ b/service/history/historyEngine.go
@@ -2864,6 +2864,7 @@ func (e *historyEngineImpl) GetReplicationMessages(
 	ctx context.Context,
 	pollingCluster string,
 	ackMessageID int64,
+	ackTimestampe time.Time,
 	queryMessageID int64,
 ) (*replicationspb.ReplicationMessages, error) {

@@ -2871,6 +2872,7 @@ func (e *historyEngineImpl) GetReplicationMessages(
 		if err := e.shard.UpdateClusterReplicationLevel(
 			pollingCluster,
 			ackMessageID,
+			ackTimestampe,
 		); err != nil {
 			e.logger.Error("error updating replication level for shard", tag.Error(err), tag.OperationFailed)
 		}
--- a/service/history/replicationTaskProcessor.go
+++ b/service/history/replicationTaskProcessor.go
@@ -87,9 +87,10 @@ type (
 		// send side
 		minTxAckedTaskID int64
 		// recv side
-		maxRxProcessedTaskID int64
-		maxRxReceivedTaskID  int64
-		rxTaskBackoff        time.Duration
+		maxRxProcessedTaskID    int64
+		maxRxProcessedTimestamp time.Time
+		maxRxReceivedTaskID     int64
+		rxTaskBackoff           time.Duration

 		requestChan   chan<- *replicationTaskRequest
 		syncShardChan chan *replicationspb.SyncShardStatus
@@ -261,6 +262,7 @@ func (p *ReplicationTaskProcessorImpl) pollProcessReplicationTasks() (retError e
 			return err
 		}
 		p.maxRxProcessedTaskID = replicationTask.GetSourceTaskId()
+		p.maxRxProcessedTimestamp = timestamp.TimeValue(replicationTask.GetVisibilityTime())
 	}

 	if !p.isStopped() {
@@ -425,9 +427,10 @@ func (p *ReplicationTaskProcessorImpl) paginationFn(_ []byte) ([]interface{}, []
 	respChan := make(chan *replicationspb.ReplicationMessages, 1)
 	p.requestChan <- &replicationTaskRequest{
 		token: &replicationspb.ReplicationToken{
-			ShardId:                p.shard.GetShardID(),
-			LastProcessedMessageId: p.maxRxProcessedTaskID,
-			LastRetrievedMessageId: p.maxRxReceivedTaskID,
+			ShardId:                     p.shard.GetShardID(),
+			LastProcessedMessageId:      p.maxRxProcessedTaskID,
+			LastProcessedVisibilityTime: &p.maxRxProcessedTimestamp,
+			LastRetrievedMessageId:      p.maxRxReceivedTaskID,
 		},
 		respChan: respChan,
 	}
--- a/service/history/replicationTaskProcessor_test.go
+++ b/service/history/replicationTaskProcessor_test.go
@@ -363,7 +363,7 @@ func (s *replicationTaskProcessorSuite) TestConvertTaskToDLQTask_History() {
 func (s *replicationTaskProcessorSuite) TestCleanupReplicationTask_Noop() {
 	ackedTaskID := int64(12345)
 	s.mockResource.ShardMgr.EXPECT().UpdateShard(gomock.Any()).Return(nil)
-	err := s.mockShard.UpdateClusterReplicationLevel(cluster.TestAlternativeClusterName, ackedTaskID)
+	err := s.mockShard.UpdateClusterReplicationLevel(cluster.TestAlternativeClusterName, ackedTaskID, time.Time{})
 	s.NoError(err)

 	s.replicationTaskProcessor.minTxAckedTaskID = ackedTaskID
@@ -374,7 +374,7 @@ func (s *replicationTaskProcessorSuite) TestCleanupReplicationTask_Noop() {
 func (s *replicationTaskProcessorSuite) TestCleanupReplicationTask_Cleanup() {
 	ackedTaskID := int64(12345)
 	s.mockResource.ShardMgr.EXPECT().UpdateShard(gomock.Any()).Return(nil)
-	err := s.mockShard.UpdateClusterReplicationLevel(cluster.TestAlternativeClusterName, ackedTaskID)
+	err := s.mockShard.UpdateClusterReplicationLevel(cluster.TestAlternativeClusterName, ackedTaskID, time.Time{})
 	s.NoError(err)

 	s.replicationTaskProcessor.minTxAckedTaskID = ackedTaskID - 1
@@ -432,13 +432,15 @@ func (s *replicationTaskProcessorSuite) TestPaginationFn_Success_More() {
 	s.replicationTaskProcessor.rxTaskBackoff = rxTaskBackoff

 	requestToken := &replicationspb.ReplicationToken{
-		ShardId:                s.shardID,
-		LastProcessedMessageId: maxRxProcessedTaskID,
-		LastRetrievedMessageId: maxRxReceivedTaskID,
+		ShardId:                     s.shardID,
+		LastProcessedMessageId:      maxRxProcessedTaskID,
+		LastProcessedVisibilityTime: &time.Time{},
+		LastRetrievedMessageId:      maxRxReceivedTaskID,
 	}

 	go func() {
 		request := <-s.requestChan
+		defer close(request.respChan)
 		s.Equal(requestToken, request.token)
 		request.respChan <- &replicationspb.ReplicationMessages{
 			SyncShardStatus:        syncShardTask,
@@ -446,7 +448,6 @@ func (s *replicationTaskProcessorSuite) TestPaginationFn_Success_More() {
 			LastRetrievedMessageId: lastRetrievedMessageID,
 			HasMore:                true,
 		}
-		close(request.respChan)
 	}()

 	tasks, _, err := s.replicationTaskProcessor.paginationFn(nil)
@@ -504,13 +505,15 @@ func (s *replicationTaskProcessorSuite) TestPaginationFn_Success_NoMore() {
 	s.replicationTaskProcessor.rxTaskBackoff = rxTaskBackoff

 	requestToken := &replicationspb.ReplicationToken{
-		ShardId:                s.shardID,
-		LastProcessedMessageId: maxRxProcessedTaskID,
-		LastRetrievedMessageId: maxRxReceivedTaskID,
+		ShardId:                     s.shardID,
+		LastProcessedMessageId:      maxRxProcessedTaskID,
+		LastProcessedVisibilityTime: &time.Time{},
+		LastRetrievedMessageId:      maxRxReceivedTaskID,
 	}

 	go func() {
 		request := <-s.requestChan
+		defer close(request.respChan)
 		s.Equal(requestToken, request.token)
 		request.respChan <- &replicationspb.ReplicationMessages{
 			SyncShardStatus:        syncShardTask,
@@ -518,7 +521,6 @@ func (s *replicationTaskProcessorSuite) TestPaginationFn_Success_NoMore() {
 			LastRetrievedMessageId: lastRetrievedMessageID,
 			HasMore:                false,
 		}
-		close(request.respChan)
 	}()

 	tasks, _, err := s.replicationTaskProcessor.paginationFn(nil)
@@ -540,15 +542,16 @@ func (s *replicationTaskProcessorSuite) TestPaginationFn_Error() {
 	s.replicationTaskProcessor.rxTaskBackoff = rxTaskBackoff

 	requestToken := &replicationspb.ReplicationToken{
-		ShardId:                s.shardID,
-		LastProcessedMessageId: maxRxProcessedTaskID,
-		LastRetrievedMessageId: maxRxReceivedTaskID,
+		ShardId:                     s.shardID,
+		LastProcessedMessageId:      maxRxProcessedTaskID,
+		LastProcessedVisibilityTime: &time.Time{},
+		LastRetrievedMessageId:      maxRxReceivedTaskID,
 	}

 	go func() {
 		request := <-s.requestChan
+		defer close(request.respChan)
 		s.Equal(requestToken, request.token)
-		close(request.respChan)
 	}()

 	tasks, _, err := s.replicationTaskProcessor.paginationFn(nil)
--- a/service/history/replicatorQueueProcessor.go
+++ b/service/history/replicatorQueueProcessor.go
@@ -386,6 +386,7 @@ func (p *replicatorQueueProcessorImpl) generateSyncActivityTask(
 						VersionHistory:     versionHistory,
 					},
 				},
+				VisibilityTime: &taskInfo.VisibilityTimestamp,
 			}, nil
 		},
 	)
@@ -455,6 +456,7 @@ func (p *replicatorQueueProcessorImpl) generateHistoryReplicationTask(
 						NewRunEvents:        newRunEventsBlob,
 					},
 				},
+				VisibilityTime: &taskInfo.VisibilityTimestamp,
 			}
 			return replicationTask, nil
 		},
--- a/service/history/replicatorQueueProcessor_test.go
+++ b/service/history/replicatorQueueProcessor_test.go
@@ -368,13 +368,14 @@ func (s *replicatorQueueProcessorSuite) TestSyncActivity_ActivityRetry() {
 	scheduleID := int64(144)
 	taskID := int64(1444)
 	version := int64(2333)
+	taskTimestamp := time.Now().UTC()
 	task := &tasks.SyncActivityTask{
 		WorkflowKey: definition.NewWorkflowKey(
 			namespaceID.String(),
 			workflowID,
 			runID,
 		),
-		VisibilityTimestamp: time.Now().UTC(),
+		VisibilityTimestamp: taskTimestamp,
 		TaskID:              taskID,
 		Version:             version,
 		ScheduledID:         scheduleID,
@@ -467,6 +468,7 @@ func (s *replicatorQueueProcessorSuite) TestSyncActivity_ActivityRetry() {
 				VersionHistory:     versionHistory,
 			},
 		},
+		VisibilityTime: timestamp.TimePtr(taskTimestamp),
 	}, result)
 }

@@ -479,13 +481,14 @@ func (s *replicatorQueueProcessorSuite) TestSyncActivity_ActivityRunning() {
 	scheduleID := int64(144)
 	taskID := int64(1444)
 	version := int64(2333)
+	taskTimestamp := time.Now().UTC()
 	task := &tasks.SyncActivityTask{
 		WorkflowKey: definition.NewWorkflowKey(
 			namespaceID.String(),
 			workflowID,
 			runID,
 		),
-		VisibilityTimestamp: time.Now().UTC(),
+		VisibilityTimestamp: taskTimestamp,
 		TaskID:              taskID,
 		Version:             version,
 		ScheduledID:         scheduleID,
@@ -580,5 +583,6 @@ func (s *replicatorQueueProcessorSuite) TestSyncActivity_ActivityRunning() {
 				VersionHistory:     versionHistory,
 			},
 		},
+		VisibilityTime: timestamp.TimePtr(taskTimestamp),
 	}, result)
 }
--- a/service/history/shard/context.go
+++ b/service/history/shard/context.go
@@ -85,7 +85,7 @@ type (
 		UpdateReplicatorDLQAckLevel(sourCluster string, ackLevel int64) error

 		GetClusterReplicationLevel(cluster string) int64
-		UpdateClusterReplicationLevel(cluster string, ackTaskID int64) error
+		UpdateClusterReplicationLevel(cluster string, ackTaskID int64, ackTimestamp time.Time) error

 		GetTimerAckLevel() time.Time
 		UpdateTimerAckLevel(ackLevel time.Time) error
--- a/service/history/shard/context_impl.go
+++ b/service/history/shard/context_impl.go
@@ -102,7 +102,13 @@ type (
 		timerMaxReadLevelMap      map[string]time.Time // cluster -> timerMaxReadLevel

 		// exist only in memory
-		remoteClusterCurrentTime map[string]time.Time
+		remoteClusterInfos map[string]*remoteClusterInfo
+	}
+
+	remoteClusterInfo struct {
+		CurrentTime               time.Time
+		AckedReplicationTaskID    int64
+		AckedReplicationTimestamp time.Time
 	}
 )

@@ -297,12 +303,14 @@ func (s *ContextImpl) GetClusterReplicationLevel(cluster string) int64 {
 	return persistence.EmptyQueueMessageID
 }

-func (s *ContextImpl) UpdateClusterReplicationLevel(cluster string, ackTaskID int64) error {
+func (s *ContextImpl) UpdateClusterReplicationLevel(cluster string, ackTaskID int64, ackTimestamp time.Time) error {
 	s.wLock()
 	defer s.wUnlock()

 	s.shardInfo.ClusterReplicationLevel[cluster] = ackTaskID
 	s.shardInfo.StolenSinceRenew = 0
+	s.getRemoteClusterInfoLocked(cluster).AckedReplicationTaskID = ackTaskID
+	s.getRemoteClusterInfoLocked(cluster).AckedReplicationTimestamp = ackTimestamp
 	return s.updateShardInfoLocked()
 }

@@ -432,7 +440,7 @@ func (s *ContextImpl) UpdateTimerMaxReadLevel(cluster string) time.Time {

 	currentTime := s.GetTimeSource().Now()
 	if cluster != "" && cluster != s.GetClusterMetadata().GetCurrentClusterName() {
-		currentTime = s.remoteClusterCurrentTime[cluster]
+		currentTime = s.getRemoteClusterInfoLocked(cluster).CurrentTime
 	}

 	s.timerMaxReadLevelMap[cluster] = currentTime.Add(s.config.TimerProcessorMaxTimeShift()).Truncate(time.Millisecond)
@@ -950,7 +958,6 @@ func (s *ContextImpl) emitShardInfoMetricsLogsLocked() {
 			logWarnTimerLevelDiff < timerLag) {

 		s.logger.Warn("Shard ack levels diff exceeds warn threshold.",
-			tag.ShardTime(s.remoteClusterCurrentTime),
 			tag.ShardReplicationAck(s.shardInfo.ReplicationAckLevel),
 			tag.ShardTimerAcks(s.shardInfo.ClusterTimerAckLevel),
 			tag.ShardTransferAcks(s.shardInfo.ClusterTransferAckLevel))
@@ -1063,9 +1070,9 @@ func (s *ContextImpl) SetCurrentTime(cluster string, currentTime time.Time) {
 	s.wLock()
 	defer s.wUnlock()
 	if cluster != s.GetClusterMetadata().GetCurrentClusterName() {
-		prevTime := s.remoteClusterCurrentTime[cluster]
+		prevTime := s.getRemoteClusterInfoLocked(cluster).CurrentTime
 		if prevTime.Before(currentTime) {
-			s.remoteClusterCurrentTime[cluster] = currentTime
+			s.getRemoteClusterInfoLocked(cluster).CurrentTime = currentTime
 		}
 	} else {
 		panic("Cannot set current time for current cluster")
@@ -1076,7 +1083,7 @@ func (s *ContextImpl) GetCurrentTime(cluster string) time.Time {
 	s.rLock()
 	defer s.rUnlock()
 	if cluster != s.GetClusterMetadata().GetCurrentClusterName() {
-		return s.remoteClusterCurrentTime[cluster]
+		return s.getRemoteClusterInfoLocked(cluster).CurrentTime
 	}
 	return s.GetTimeSource().Now().UTC()
 }
@@ -1394,7 +1401,7 @@ func (s *ContextImpl) loadShardMetadata(ownershipChanged *bool) error {
 	updatedShardInfo.Owner = s.GetHostInfo().Identity()

 	// initialize the cluster current time to be the same as ack level
-	remoteClusterCurrentTime := make(map[string]time.Time)
+	remoteClusterInfos := make(map[string]*remoteClusterInfo)
 	timerMaxReadLevelMap := make(map[string]time.Time)
 	for clusterName, info := range s.GetClusterMetadata().GetAllClusterInfo() {
 		if !info.Enabled {
@@ -1407,7 +1414,7 @@ func (s *ContextImpl) loadShardMetadata(ownershipChanged *bool) error {
 				currentReadTime = timestamp.TimeValue(currentTime)
 			}

-			remoteClusterCurrentTime[clusterName] = currentReadTime
+			remoteClusterInfos[clusterName] = &remoteClusterInfo{CurrentTime: currentReadTime}
 			timerMaxReadLevelMap[clusterName] = currentReadTime
 		} else { // active cluster
 			timerMaxReadLevelMap[clusterName] = currentReadTime
@@ -1424,12 +1431,23 @@ func (s *ContextImpl) loadShardMetadata(ownershipChanged *bool) error {
 	}

 	s.shardInfo = updatedShardInfo
-	s.remoteClusterCurrentTime = remoteClusterCurrentTime
+	s.remoteClusterInfos = remoteClusterInfos
 	s.timerMaxReadLevelMap = timerMaxReadLevelMap

 	return nil
 }

+func (s *ContextImpl) getRemoteClusterInfoLocked(clusterName string) *remoteClusterInfo {
+	if info, ok := s.remoteClusterInfos[clusterName]; ok {
+		return info
+	}
+	info := &remoteClusterInfo{
+		AckedReplicationTaskID: persistence.EmptyQueueMessageID,
+	}
+	s.remoteClusterInfos[clusterName] = info
+	return info
+}
+
 func (s *ContextImpl) acquireShard() {
 	// Retry for 5m, with interval up to 10s (default)
 	policy := backoff.NewExponentialRetryPolicy(50 * time.Millisecond)
--- a/service/history/shard/context_mock.go
+++ b/service/history/shard/context_mock.go
@@ -592,17 +592,17 @@ func (mr *MockContextMockRecorder) SetCurrentTime(cluster, currentTime interface
 }

 // UpdateClusterReplicationLevel mocks base method.
-func (m *MockContext) UpdateClusterReplicationLevel(cluster string, ackTaskID int64) error {
+func (m *MockContext) UpdateClusterReplicationLevel(cluster string, ackTaskID int64, ackTimestamp time.Time) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UpdateClusterReplicationLevel", cluster, ackTaskID)
+	ret := m.ctrl.Call(m, "UpdateClusterReplicationLevel", cluster, ackTaskID, ackTimestamp)
 	ret0, _ := ret[0].(error)
 	return ret0
 }

 // UpdateClusterReplicationLevel indicates an expected call of UpdateClusterReplicationLevel.
-func (mr *MockContextMockRecorder) UpdateClusterReplicationLevel(cluster, ackTaskID interface{}) *gomock.Call {
+func (mr *MockContextMockRecorder) UpdateClusterReplicationLevel(cluster, ackTaskID, ackTimestamp interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateClusterReplicationLevel", reflect.TypeOf((*MockContext)(nil).UpdateClusterReplicationLevel), cluster, ackTaskID)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateClusterReplicationLevel", reflect.TypeOf((*MockContext)(nil).UpdateClusterReplicationLevel), cluster, ackTaskID, ackTimestamp)
 }

 // UpdateNamespaceNotificationVersion mocks base method.
--- a/service/history/shard/context_testutil.go
+++ b/service/history/shard/context_testutil.go
@@ -68,7 +68,7 @@ func NewTestContext(
 		transferMaxReadLevel:      0,
 		maxTransferSequenceNumber: 100000,
 		timerMaxReadLevelMap:      make(map[string]time.Time),
-		remoteClusterCurrentTime:  make(map[string]time.Time),
+		remoteClusterInfos:        make(map[string]*remoteClusterInfo),
 	}
 	return &ContextTest{
 		ContextImpl:     shard,
--- a/service/history/shard/engine.go
+++ b/service/history/shard/engine.go
@@ -28,6 +28,7 @@ package shard

 import (
 	"context"
+	"time"

 	commonpb "go.temporal.io/api/common/v1"
 	historypb "go.temporal.io/api/history/v1"
@@ -69,7 +70,7 @@ type (
 		ReplicateEventsV2(ctx context.Context, request *historyservice.ReplicateEventsV2Request) error
 		SyncShardStatus(ctx context.Context, request *historyservice.SyncShardStatusRequest) error
 		SyncActivity(ctx context.Context, request *historyservice.SyncActivityRequest) error
-		GetReplicationMessages(ctx context.Context, pollingCluster string, ackMessageID int64, queryMessageID int64) (*replicationspb.ReplicationMessages, error)
+		GetReplicationMessages(ctx context.Context, pollingCluster string, ackMessageID int64, ackTimestamp time.Time, queryMessageID int64) (*replicationspb.ReplicationMessages, error)
 		GetDLQReplicationMessages(ctx context.Context, taskInfos []*replicationspb.ReplicationTaskInfo) ([]*replicationspb.ReplicationTask, error)
 		QueryWorkflow(ctx context.Context, request *historyservice.QueryWorkflowRequest) (*historyservice.QueryWorkflowResponse, error)
 		ReapplyEvents(ctx context.Context, namespaceUUID namespace.ID, workflowID string, runID string, events []*historypb.HistoryEvent) error
--- a/service/history/shard/engine_mock.go
+++ b/service/history/shard/engine_mock.go
@@ -31,6 +31,7 @@ package shard
 import (
 	context "context"
 	reflect "reflect"
+	time "time"

 	gomock "github.com/golang/mock/gomock"
 	common "go.temporal.io/api/common/v1"
@@ -156,18 +157,18 @@ func (mr *MockEngineMockRecorder) GetMutableState(ctx, request interface{}) *gom
 }

 // GetReplicationMessages mocks base method.
-func (m *MockEngine) GetReplicationMessages(ctx context.Context, pollingCluster string, ackMessageID, queryMessageID int64) (*repication.ReplicationMessages, error) {
+func (m *MockEngine) GetReplicationMessages(ctx context.Context, pollingCluster string, ackMessageID int64, ackTimestamp time.Time, queryMessageID int64) (*repication.ReplicationMessages, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetReplicationMessages", ctx, pollingCluster, ackMessageID, queryMessageID)
+	ret := m.ctrl.Call(m, "GetReplicationMessages", ctx, pollingCluster, ackMessageID, ackTimestamp, queryMessageID)
 	ret0, _ := ret[0].(*repication.ReplicationMessages)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }

 // GetReplicationMessages indicates an expected call of GetReplicationMessages.
-func (mr *MockEngineMockRecorder) GetReplicationMessages(ctx, pollingCluster, ackMessageID, queryMessageID interface{}) *gomock.Call {
+func (mr *MockEngineMockRecorder) GetReplicationMessages(ctx, pollingCluster, ackMessageID, ackTimestamp, queryMessageID interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetReplicationMessages", reflect.TypeOf((*MockEngine)(nil).GetReplicationMessages), ctx, pollingCluster, ackMessageID, queryMessageID)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetReplicationMessages", reflect.TypeOf((*MockEngine)(nil).GetReplicationMessages), ctx, pollingCluster, ackMessageID, ackTimestamp, queryMessageID)
 }

 // MergeDLQMessages mocks base method.