We mark the source as idle based on this check:

if (!changeEvent && !eventResumeToken) {
  _isIdle.store(true);        
  return false;    
}

However, eventResumeToken will always be non-null since the underlying _change_stream_ object will always return the last resumeToken it has.

This check needs to be enhanced so that it does a comparison between the currently obtained resumeToken and the previously obtained value and that decides if we have a "new" resumeToken or not.