fix worker failed to commit offset problem for federated mode on route re-assign
refactor on worker federated mode
3.1. throw exception on unexpected behaviour
3.2. add more check to avoid potential race condition
3.3. add unit test for error scenarios
not to shutdown fetcher thread when the assignment is empty