Close ChangeStream after non-resumable non-timeout errors

(venv) ➜  motor git:(MOTOR-951) ✗ git -C ../mongo-python-driver checkout 935f926bd9bf556cadba5d7bda344371b4da24ef                                              

HEAD is now at 935f926b PYTHON-3362 Ignore wtimeout when timeoutMS or timeout() is configured (#1013)

(venv) ➜  motor git:(MOTOR-951) ✗ rm -rf .eggs && python setup.py test -s test.asyncio_tests.test_asyncio_change_stream.TestAsyncIOChangeStream.test_missing_id

...

test_missing_id (test.asyncio_tests.test_asyncio_change_stream.TestAsyncIOChangeStream) ... ok

----------------------------------------------------------------------

Ran 1 test in 0.641s

OK

(venv) ➜  motor git:(MOTOR-951) ✗ git -C ../mongo-python-driver checkout 4d4fddaf699d16af6e082da5b5c3303cbafc2818                                              

Previous HEAD position was 935f926b PYTHON-3362 Ignore wtimeout when timeoutMS or timeout() is configured (#1013)

HEAD is now at 4d4fddaf PYTHON-3363 Allow change stream to be resumed after a timeout (#1014)

(venv) ➜  motor git:(MOTOR-951) ✗ rm -rf .eggs && python setup.py test -s test.asyncio_tests.test_asyncio_change_stream.TestAsyncIOChangeStream.test_missing_id

...

test_missing_id (test.asyncio_tests.test_asyncio_change_stream.TestAsyncIOChangeStream) ... ERROR

======================================================================

ERROR: test_missing_id (test.asyncio_tests.test_asyncio_change_stream.TestAsyncIOChangeStream)

----------------------------------------------------------------------

Traceback (most recent call last):

  File "/Users/julius/Work/motor/test/asyncio_tests/__init__.py", line 49, in __call__

    result = self.orig_method()

  File "/Users/julius/Work/motor/test/asyncio_tests/__init__.py", line 209, in wrapped

    self.loop.run_until_complete(task)

  File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete

    return future.result()

  File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/asyncio/tasks.py", line 494, in wait_for

    return fut.result()

  File "/Users/julius/Work/motor/test/asyncio_tests/test_asyncio_change_stream.py", line 194, in test_missing_id

    await change_stream.next()

  File "/Users/julius/Work/motor/motor/core.py", line 1838, in next

    doc = await self.try_next()

  File "/Users/julius/Work/motor/motor/core.py", line 1879, in try_next

    return await self._framework.run_on_executor(loop, self._try_next)

  File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/concurrent/futures/thread.py", line 57, in run

    result = self.fn(*self.args, **self.kwargs)

  File "/Users/julius/Work/motor/motor/core.py", line 1784, in _try_next

    return self.delegate.try_next()

  File "/Users/julius/Work/mongo-python-driver/pymongo/_csot.py", line 105, in csot_wrapper

    return func(self, *args, **kwargs)

  File "/Users/julius/Work/mongo-python-driver/pymongo/change_stream.py", line 341, in try_next

    self._resume()

  File "/Users/julius/Work/mongo-python-driver/pymongo/change_stream.py", line 235, in _resume

    self._cursor = self._create_cursor()

  File "/Users/julius/Work/mongo-python-driver/pymongo/change_stream.py", line 227, in _create_cursor

    return self._run_aggregation_cmd(session=s, explicit_session=self._session is not None)

  File "/Users/julius/Work/mongo-python-driver/pymongo/change_stream.py", line 221, in _run_aggregation_cmd

    return self._client._retryable_read(

  File "/Users/julius/Work/mongo-python-driver/pymongo/_csot.py", line 105, in csot_wrapper

    return func(self, *args, **kwargs)

  File "/Users/julius/Work/mongo-python-driver/pymongo/mongo_client.py", line 1420, in _retryable_read

    return func(session, server, sock_info, read_pref)

  File "/Users/julius/Work/mongo-python-driver/pymongo/aggregation.py", line 140, in get_cursor

    result = sock_info.command(

  File "/Users/julius/Work/mongo-python-driver/pymongo/pool.py", line 766, in command

    return command(

  File "/Users/julius/Work/mongo-python-driver/pymongo/network.py", line 166, in command

    helpers._check_command_response(

  File "/Users/julius/Work/mongo-python-driver/pymongo/helpers.py", line 181, in _check_command_response

    raise OperationFailure(errmsg, code, response, max_wire_version)

pymongo.errors.OperationFailure: PlanExecutor error during aggregation :: caused by :: Encountered an event whose _id field, which contains the resume token, was modified by the pipeline. Modifying the _id field of an event makes it impossible to resume the stream from that point. Only transformations that retain the unmodified _id field are allowed. Expected: { _id: { _data: "8262EA9F630000006D2B022C0100296E5A100453CFFF4695794B4FB46900ADF0F3661346645F6964006462EA8D394DC02F7506531FB70004" } } but found: {}, full error: {'errorLabels': ['NonResumableChangeStreamError'], 'ok': 0.0, 'errmsg': 'PlanExecutor error during aggregation :: caused by :: Encountered an event whose _id field, which contains the resume token, was modified by the pipeline. Modifying the _id field of an event makes it impossible to resume the stream from that point. Only transformations that retain the unmodified _id field are allowed. Expected: { _id: { _data: "8262EA9F630000006D2B022C0100296E5A100453CFFF4695794B4FB46900ADF0F3661346645F6964006462EA8D394DC02F7506531FB70004" } } but found: {}', 'code': 280, 'codeName': 'ChangeStreamFatalError', '$clusterTime': {'clusterTime': Timestamp(1659543395, 109), 'signature': {'hash': b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', 'keyId': 0}}, 'operationTime': Timestamp(1659543395, 109)}

----------------------------------------------------------------------

Ran 1 test in 0.705s

FAILED (errors=1)