Currently the split fetchers are only removed from the SplitFetcherManager.fetchers when the thread exit. The may cause problem because when SplitFetcherManager.addSplits() is called, it may see a shutting down split fetcher and adds splits to it. These splits will then just be lost.
This issue is actually already fixed in
FLINK-18128. The fix needs to cherry-picked to 1.11.3