title: Reactive mongo change streams is resuming with an invalid resume token
body: <p>Using the reactive mongo template, I am trying to listen to the mongo change streams for 6 collections for the insert operations only. My code to start listening to change streams is below:</p>
  
  <p>Change streams initiation:</p>
  
  <pre><code>//Loop for 6 collections
  reactiveMongoTemplate.changeStream(collectionName, changeStreamOptions, MongoChangeStreamEvent.class)
            .doOnNext(changeStreamEvent -&gt; {          
              System.out.println("In doOnNext: Received a new change stream event ");         
            })
            .map(changeStreamEvent -&gt; {    
              //save resume token
            })
            .onErrorResume(throwable -&gt; {
              System.out.println("In onErrorResume of change stream event :: " + throwable.getMessage());
              return null;
            })
            .subscribe();
  </code></pre>
  
  <p>On each new entry, I save the resume token in one of my collections:</p>
  
  <pre><code>//Document
  public class MongoChangeStreamEvent implements Serializable {
    @Id
    private String id;
    private String resumeToken;
    //other fields and getters and setters
  }
  </code></pre>
  
  <p>Used change stream options for initiation.
  If a resume token exists in the collection, use it.
  Else, resume right now.</p>
  
  <pre><code>
  ChangeStreamOptions changeStreamOptions;
      final BsonDocument resumeToken = //getLatestResumeToken, first entry after sorting 
                                       //MongoChangeStreamEvent collection in descending order based on 
                                       //resumeToken
  
      if (!resumeToken.isEmpty()) {
        changeStreamOptions = ChangeStreamOptions.builder()
            .filter(newAggregation(match(where(OPERATION_TYPE).is(INSERT_OPERATION))))
            .resumeAfter(resumeToken)
            .build();
      } else {
        changeStreamOptions = ChangeStreamOptions.builder()
            .filter(newAggregation(match(where(OPERATION_TYPE).is(INSERT_OPERATION))))        
            .resumeAt(Instant.now())
            .build();
      }
  
  </code></pre>
  
  <p>Error thrown:</p>
  
  <pre><code>Suppressed: com.mongodb.MongoQueryException: Query failed with error code 280 and error message 
  'cannot resume stream; the resume token was not found. {_data: 
  "825EAFCA12000000162B022C0100296E5A100440284C0343E64ADEB43522FC0552CC1446645F696400645EAFCA12B51E93000716B9300004"}'
  </code></pre>
  
  <p>So far, I have had different experiences with the working of this functionality. Application starting without an existing resume token was always working as expected. Other findings when I had restarted the application with existing resume token are below:</p>
  
  <ol>
  <li>It sometimes worked perfectly for all the 6.</li>
  <li>Just after the restart, it initiated for a few and failed for the remaining.</li>
  <li>Just after the restart, initiation didn't throw any error. But on insertion of a document in the collection being watched, few/all errored out.</li>
  </ol>
  
  <p>I understand that the change stream is dependent on the oplogs history as quoted in the <a href="https://docs.mongodb.com/manual/changeStreams/#resume-a-change-stream" rel="nofollow noreferrer">documentation</a>. What's more surprising to me was the resume token that gets errored out didn't match any of my existing resume tokens and wasn't present in the oplogs too. </p>
  
  <p>I have verified that the resume token submitted to the reactive mongo template was always correct. </p>
  
  <ol>
  <li>Please let me know if I am missing anything. </li>
  <li>Also, I would love to know how to handle the failure of one/a few out of many change streams running.</li>
  </ol>