Shard end v2

[ychunxue]

Description of changes:

Change is made in ShutdownTask. When reaching a ShutdownTask with Shard_End shutdown reason, do a re-validation about if the current shard reached shard_end. If it didn't reach the shard_end, force the current ShardConsumer to lose the lease and shutdown with Lease_Lost state. This would enable other workers to pick up the lease for the unclosed shard.

Test did for the changes:

Unit tests covered ShardConsumer shutdown in three different scenarios: Lease lost, ShardEnd and incorrect ShardEnd.

Functional test done in real ShardEnd and incorrect ShardEnd scenarios.

Long running regression test is done with Split and Merge accounts. And also having the incorrect ShardEnd triggered randomly during the test to catch regressions.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.