1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
|
/**
* Test basic, steady-state replication change stream functionality with speculative majority reads.
*
* @tags: [uses_speculative_majority]
*/
(function() {
"use strict";
load("jstests/libs/write_concern_util.js"); // for [stop|restart]ServerReplication.
const name = "change_stream_speculative_majority";
const replTest = new ReplSetTest({
name: name,
nodes: [{}, {rsConfig: {priority: 0}}],
nodeOptions: {enableMajorityReadConcern: 'false'}
});
replTest.startSet();
replTest.initiate();
const dbName = name;
const collName = "coll";
let primary = replTest.getPrimary();
let secondary = replTest.getSecondary();
let primaryDB = primary.getDB(dbName);
let primaryColl = primaryDB[collName];
// Open a change stream.
let res = primaryDB.runCommand(
{aggregate: collName, pipeline: [{$changeStream: {}}], cursor: {}, maxTimeMS: 5000});
assert.commandWorked(res);
let cursorId = res.cursor.id;
// Insert a document on primary and let it majority commit.
assert.commandWorked(primaryColl.insert({_id: 1}, {writeConcern: {w: "majority"}}));
// Receive the first change event.
res = primary.getDB(dbName).runCommand({getMore: cursorId, collection: collName});
let changes = res.cursor.nextBatch;
assert.eq(changes.length, 1);
assert.eq(changes[0]["fullDocument"], {_id: 1});
assert.eq(changes[0]["operationType"], "insert");
// Save the resume token.
let resumeToken = changes[0]["_id"];
// This query should time out waiting for new results and return an empty batch.
res = primary.getDB(dbName).runCommand({getMore: cursorId, collection: collName, maxTimeMS: 5000});
assert.eq(res.cursor.nextBatch, []);
// Pause replication on the secondary so that writes won't majority commit.
stopServerReplication(secondary);
// Do a new write on primary.
assert.commandWorked(primaryColl.insert({_id: 2}));
// The change stream query should time out waiting for the new result to majority commit.
res = primary.getDB(dbName).runCommand({getMore: cursorId, collection: collName, maxTimeMS: 5000});
assert.commandFailedWithCode(res, ErrorCodes.MaxTimeMSExpired);
// An aggregate trying to resume a stream that includes the change should also time out.
res = primaryDB.runCommand({
aggregate: collName,
pipeline: [{$changeStream: {resumeAfter: resumeToken}}],
cursor: {},
maxTimeMS: 5000
});
assert.commandFailedWithCode(res, ErrorCodes.MaxTimeMSExpired);
// Resume the stream after restarting replication. We should now be able to see the new event.
restartServerReplication(secondary);
replTest.awaitReplication();
// Re-open the stream, and receive the new event.
res = primaryDB.runCommand(
{aggregate: collName, pipeline: [{$changeStream: {resumeAfter: resumeToken}}], cursor: {}});
assert.commandWorked(res);
changes = res.cursor.firstBatch;
assert.eq(changes.length, 1);
assert.eq(changes[0]["fullDocument"], {_id: 2});
assert.eq(changes[0]["operationType"], "insert");
replTest.stopSet();
})();
|
