408
}, nil
409
411
>
taskAttributes := replicationTask.GetHistoryTaskAttributes()
412
>
413
>
events, err := p.historySerializer.DeserializeEvents(taskAttributes.GetEvents())
414
>
if err != nil {
415
return nil, err
416
}
417
419
p.logger.Error("Empty events in a batch")
420
return nil, ErrCorruptedHistoryEventBatch
421
}
423
>
lastEvent := events[len(events)-1]
424
>
// NOTE: last event vs next event, next event ID is exclusive
425
>
nextEventID := lastEvent.GetEventId() + 1
426
>
427
>
return &persistence.PutReplicationTaskToDLQRequest{
428
>
SourceClusterName: p.sourceCluster,
429
>
TaskInfo: &persistencespb.ReplicationTaskInfo{
430
>
NamespaceId: taskAttributes.GetNamespaceId(),
431
>
WorkflowId: taskAttributes.GetWorkflowId(),
432
>
RunId: taskAttributes.GetRunId(),
433
>
TaskId: replicationTask.GetSourceTaskId(),
434
>
TaskType: enumsspb.TASK_TYPE_REPLICATION_HISTORY,
435
>
FirstEventId: firstEvent.GetEventId(),
436
>
NextEventId: nextEventID,
437
>
Version: firstEvent.GetVersion(),
438
>
VisibilityTime: replicationTask.GetVisibilityTime(),
439
>
NewRunId: taskAttributes.GetNewRunId(),
440
>
// BranchToken & NewRunBranchToken should also be populated but are deprecated
441
>
},
442
>
}, nil
443
444
case enumsspb.REPLICATION_TASK_TYPE_SYNC_WORKFLOW_STATE_TASK: