jstests/multiVersion/genericChangeStreams/change_streams_read_oplog_after_downgrade.js
/**
 * Validates that resuming each of the change stream will not crash the server. The 'changeStreams'
 * should be an array and each entry should have fields 'watch', 'resumeToken' and
 * 'endSentinelEntry'.
 */
Validates that resuming each of the change stream will not crash the server.
Validates that resuming each of the change stream will not crash the server .
function resumeStreamsOnDowngradedVersion(changeStreams) {
    for (let changeStream of changeStreams) {
        jsTestLog("Validating change stream for " + tojson(changeStream));
        const csCursor = changeStream.watch({resumeAfter: changeStream.resumeToken});
        assert.soon(() => {
            if (!csCursor.hasNext()) {
                return false;
            }
            try {
                const nextEvent = csCursor.next();
                return (nextEvent.documentKey._id == changeStream.endSentinelEntry);
            } catch (e) {
                jsTestLog("Error occurred while reading change stream. " + tojson(e));
                assert.commandWorked(shardedColl.runCommand({ping: 1}));
                assert.commandWorked(st.rs0.getPrimary().getDB("admin").runCommand({ping: 1}));
                assert.commandWorked(st.configRS.getPrimary().getDB("admin").runCommand({ping: 1}));
                return true;
            }
        });
    }
}
function resumeStreamsOnDowngradedVersion ( changeStreams ) { for ( let changeStream of changeStreams ) { jsTestLog ( "Validating change stream for " + tojson ( changeStream ) ) ; const csCursor = changeStream . watch ( { resumeAfter : changeStream . resumeToken } ) ; assert . soon ( ( ) => { if ( ! csCursor . hasNext ( ) ) { return false ; } try { const nextEvent = csCursor . next ( ) ; return ( nextEvent . documentKey . _id == changeStream . endSentinelEntry ) ; } catch ( e ) { jsTestLog ( "Error occurred while reading change stream. " + tojson ( e ) ) ; assert . commandWorked ( shardedColl . runCommand ( { ping : 1 } ) ) ; assert . commandWorked ( st . rs0 . getPrimary ( ) . getDB ( "admin" ) . runCommand ( { ping : 1 } ) ) ; assert . commandWorked ( st . configRS . getPrimary ( ) . getDB ( "admin" ) . runCommand ( { ping : 1 } ) ) ; return true ; } } ) ; } }
Validates that resuming each of the change stream will not crash the server.
Validates that resuming each of the change stream will not crash the server .
// Keep calling 'getmore' until the sentinal entry for the next test is found or until the // change stream throws an error. // Validate that the error returned was not a consequence of server crash.