62
serializer serialization.Serializer,
63
logger log.Logger,
65
>
testSer := &testSerializer{Serializer: serializer}
66
>
return &ExecutionMutableStateTaskSuite{
67
>
Assertions: require.New(t),
68
>
ShardManager: p.NewShardManager(
69
>
shardStore,
70
>
testSer,
71
>
),
72
>
ExecutionManager: p.NewExecutionManager(
73
>
executionStore,
74
>
testSer,
75
>
nil,
76
>
logger,
77
>
dynamicconfig.GetIntPropertyFn(4*1024*1024),
78
>
dynamicconfig.GetBoolPropertyFn(false),
79
>
),
80
>
Logger: logger,
81
>
}
82
>
}
83
85
>
s.Assertions = require.New(s.T())
86
>
s.Ctx, s.Cancel = context.WithTimeout(context.Background(), 30*time.Second*debug.TimeoutMultiplier)
87
>
88
>
s.ShardID++
89
>
resp, err := s.ShardManager.GetOrCreateShard(s.Ctx, &p.GetOrCreateShardRequest{
90
>
ShardID: s.ShardID,
91
>
InitialShardInfo: &persistencespb.ShardInfo{
92
>
ShardId: s.ShardID,
93
>
RangeId: 1,
94
>
Owner: "test-shard-owner",
95
>
},
96
>
})
97
>
s.NoError(err)
98
>
previousRangeID := resp.ShardInfo.RangeId
99
>
resp.ShardInfo.RangeId++
100
>
err = s.ShardManager.UpdateShard(s.Ctx, &p.UpdateShardRequest{
101
>
ShardInfo: resp.ShardInfo,
102
>
PreviousRangeID: previousRangeID,
103
>
})
104
>
s.NoError(err)
105
>
s.RangeID = resp.ShardInfo.RangeId
106
>
s.Owner = resp.ShardInfo.Owner
107
>
108
>
s.WorkflowKey = definition.NewWorkflowKey(
109
>
uuid.New().String(),
110
>
uuid.New().String(),
111
>
uuid.New().String(),
112
>
)
113
>
}
114
116
>
for _, category := range []tasks.Category{tasks.CategoryTransfer, tasks.CategoryReplication, tasks.CategoryVisibility} {
117
>
err := s.ExecutionManager.RangeCompleteHistoryTasks(s.Ctx, &p.RangeCompleteHistoryTasksRequest{
118
>
ShardID: s.ShardID,
119
>
TaskCategory: category,
120
>
InclusiveMinTaskKey: tasks.NewImmediateKey(0),
121
>
ExclusiveMaxTaskKey: tasks.NewImmediateKey(math.MaxInt64),
122
>
})
123
>
s.NoError(err)
124
>
}
125
>
err := s.ExecutionManager.RangeCompleteHistoryTasks(s.Ctx, &p.RangeCompleteHistoryTasksRequest{
126
>
ShardID: s.ShardID,
127
>
TaskCategory: tasks.CategoryTimer,
128
>
InclusiveMinTaskKey: tasks.NewKey(time.Unix(0, 0), 0),
129
>
ExclusiveMaxTaskKey: tasks.NewKey(time.Unix(0, math.MaxInt64), 0),
130
>
})
131
>
s.NoError(err)
132
>
133
>
s.Cancel()
134
}
135