151
return err
152
}
154
>
err = multierr.Append(err, outputFile.Close())
155
>
}()
156
>
adminClient := ac.clientFactory.AdminClient(c)
157
>
158
>
pageSize := c.Int(FlagPageSize)
159
>
iterator := collection.NewPagingIterator[*commonspb.HistoryDLQTask](
160
>
func(paginationToken []byte) ([]*commonspb.HistoryDLQTask, []byte, error) {
161
>
request := &adminservice.GetDLQTasksRequest{
162
>
DlqKey: &commonspb.HistoryDLQKey{
163
>
TaskCategory: int32(ac.category.ID()),
164
>
SourceCluster: ac.sourceCluster,
165
>
TargetCluster: ac.targetCluster,
166
>
},
167
>
PageSize: int32(pageSize),
168
>
NextPageToken: paginationToken,
169
>
}
170
>
res, err := adminClient.GetDLQTasks(ctx, request)
171
>
if err != nil {
172
// If the DLQ does not exist yet, it's effectively empty, so we can safely return without an error.
173
if strings.Contains(err.Error(), "queue not found:") {