Java - Change Streams

1
git clone https://github.com/mongodb-developer/java-quick-start

1
MongoCollection<Grade> grades = db.getCollection("grades", Grade.class);
2
ChangeStreamIterable<Grade> changeStream = grades.watch();
3
changeStream.forEach((Consumer<ChangeStreamDocument<Grade>>) System.out::println);

1
grades.watch().forEach(printEvent());
2
3
private static Consumer<ChangeStreamDocument<Grade>> printEvent() {
4
    return System.out::println;
5
}

1
mvn compile exec:java -Dexec.mainClass="com.mongodb.quickstart.ChangeStreams" -Dmongodb.uri="mongodb+srv://USERNAME:PASSWORD@cluster0-abcde.mongodb.net/test?w=majority"

1
mvn compile exec:java -Dexec.mainClass="com.mongodb.quickstart.MappingPOJO" -Dmongodb.uri="mongodb+srv://USERNAME:PASSWORD@cluster0-abcde.mongodb.net/test?w=majority"

1
Grade inserted.
2
Grade found:    Grade{id=5e2b4a28c9e9d55e3d7dbacf, student_id=10003.0, class_id=10.0, scores=[Score{type='homework', score=50.0}]}
3
Grade replaced: Grade{id=5e2b4a28c9e9d55e3d7dbacf, student_id=10003.0, class_id=10.0, scores=[Score{type='homework', score=50.0}, Score{type='exam', score=42.0}]}
4
Grade deleted:  AcknowledgedDeleteResult{deletedCount=1}

1
ChangeStreamDocument{
2
   operationType=OperationType{ value='insert' },
3
   resumeToken={ "_data":"825E2F3E40000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F3E400C47CF19D59361620004" },
4
   namespace=sample_training.grades,
5
   destinationNamespace=null,
6
   fullDocument=Grade{
7
      id=5e2f3e400c47cf19d5936162,
8
      student_id=10003.0,
9
      class_id=10.0,
10
      scores=[ Score { type='homework', score=50.0 } ]
11
   },
12
   documentKey={ "_id":{ "$oid":"5e2f3e400c47cf19d5936162" } },
13
   clusterTime=Timestamp{
14
      value=6786711608069455873,
15
      seconds=1580154432,
16
      inc=1
17
   },
18
   updateDescription=null,
19
   txnNumber=null,
20
   lsid=null
21
}
22
ChangeStreamDocument{ operationType=OperationType{ value= 'replace' },
23
   resumeToken={ "_data":"825E2F3E40000000032B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F3E400C47CF19D59361620004" },
24
   namespace=sample_training.grades,
25
   destinationNamespace=null,
26
   fullDocument=Grade{
27
      id=5e2f3e400c47cf19d5936162,
28
      student_id=10003.0,
29
      class_id=10.0,
30
      scores=[ Score{ type='homework', score=50.0 }, Score{ type='exam', score=42.0 } ]
31
   },
32
   documentKey={ "_id":{ "$oid":"5e2f3e400c47cf19d5936162" } },
33
   clusterTime=Timestamp{
34
      value=6786711608069455875,
35
      seconds=1580154432,
36
      inc=3
37
   },
38
   updateDescription=null,
39
   txnNumber=null,
40
   lsid=null
41
}
42
ChangeStreamDocument{
43
   operationType=OperationType{ value='delete' },
44
   resumeToken={ "_data":"825E2F3E40000000042B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F3E400C47CF19D59361620004" },
45
   namespace=sample_training.grades,
46
   destinationNamespace=null,
47
   fullDocument=null,
48
   documentKey={ "_id":{ "$oid":"5e2f3e400c47cf19d5936162" } },
49
   clusterTime=Timestamp{
50
      value=6786711608069455876,
51
      seconds=1580154432,
52
      inc=4
53
   },
54
   updateDescription=null,
55
   txnNumber=null,
56
   lsid=null
57
}

1
List<Bson> pipeline = List.of(match(in("operationType", List.of("insert", "delete"))));
2
grades.watch(pipeline).forEach(printEvent());

1
ChangeStreamDocument {operationType=OperationType {value= 'insert'},
2
  resumeToken= {"_data": "825E2F4983000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F4983CC1D2842BFF555640004"},
3
  namespace=sample_training.grades,
4
  destinationNamespace=null,
5
  fullDocument=Grade
6
  {
7
    id=5e2f4983cc1d2842bff55564,
8
    student_id=10003.0,
9
    class_id=10.0,
10
    scores= [ Score {type= 'homework', score=50.0}]
11
  },
12
  documentKey= {"_id": {"$oid": "5e2f4983cc1d2842bff55564" }},
13
  clusterTime=Timestamp {value=6786723990460170241, seconds=1580157315, inc=1 },
14
  updateDescription=null,
15
  txnNumber=null,
16
  lsid=null
17
}
18
19
ChangeStreamDocument { operationType=OperationType {value= 'delete'},
20
  resumeToken= {"_data": "825E2F4983000000042B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E2F4983CC1D2842BFF555640004"},
21
  namespace=sample_training.grades,
22
  destinationNamespace=null,
23
  fullDocument=null,
24
  documentKey= {"_id": {"$oid": "5e2f4983cc1d2842bff55564"}},
25
  clusterTime=Timestamp {value=6786723990460170244, seconds=1580157315, inc=4},
26
  updateDescription=null,
27
  txnNumber=null,
28
  lsid=null
29
  }
30
]

1
List<Bson> pipeline = List.of(match(eq("operationType", "update")));
2
grades.watch(pipeline).forEach(printEvent());

1
ChangeStreamDocument {operationType=OperationType {value= 'update'},
2
  resumeToken= {"_data": "825E2FB83E000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCCE74AA51A0486763FE0004"},
3
  namespace=sample_training.grades,
4
  destinationNamespace=null,
5
  fullDocument=null,
6
  documentKey= {"_id": {"$oid": "5e27bcce74aa51a0486763fe"}},
7
  clusterTime=Timestamp {value=6786845739898109953, seconds=1580185662, inc=1},
8
  updateDescription=UpdateDescription {removedFields= [], updatedFields= {"comments.10": "You will learn a lot if you read the MongoDB blog!"}},
9
  txnNumber=null,
10
  lsid=null
11
}

1
List<Bson> pipeline = List.of(match(eq("operationType", "update")));
2
grades.watch(pipeline).fullDocument(UPDATE_LOOKUP).forEach(printEvent());

1
ChangeStreamDocument {operationType=OperationType {value= 'update'},
2
  resumeToken= {"_data": "825E2FBBC1000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCCE74AA51A0486763FE0004"},
3
  namespace=sample_training.grades,
4
  destinationNamespace=null,
5
  fullDocument=Grade
6
    {
7
      id=5e27bcce74aa51a0486763fe,
8
      student_id=10002.0,
9
      class_id=10.0,
10
      scores=null
11
    },
12
  documentKey= {"_id": {"$oid": "5e27bcce74aa51a0486763fe" }},
13
  clusterTime=Timestamp {value=6786849601073709057, seconds=1580186561, inc=1 },
14
  updateDescription=UpdateDescription {removedFields= [], updatedFields= {"comments.11": "You will learn a lot if you read the MongoDB blog!"}},
15
  txnNumber=null,
16
  lsid=null
17
}

1
MongoCollection<Document> grades = db.getCollection("grades");
2
List<Bson> pipeline = List.of(match(eq("operationType", "update")));
3
grades.watch(pipeline).fullDocument(UPDATE_LOOKUP).forEach((Consumer<ChangeStreamDocument<Document>>) System.out::println);

1
ChangeStreamDocument {operationType=OperationType {value= 'update'},
2
  resumeToken= {"_data": "825E2FBD89000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCCE74AA51A0486763FE0004"},
3
  namespace=sample_training.grades,
4
  destinationNamespace=null,
5
  fullDocument=Document {
6
    {
7
      _id=5e27bcce74aa51a0486763fe,
8
      class_id=10.0,
9
      student_id=10002.0,
10
      comments= [ You will learn a lot if you read the MongoDB blog!, [...], You will learn a lot if you read the MongoDB blog!]
11
    }
12
  },
13
  documentKey= {"_id": {"$oid": "5e27bcce74aa51a0486763fe"}},
14
  clusterTime=Timestamp {value=6786851559578796033, seconds=1580187017, inc=1},
15
  updateDescription=UpdateDescription {removedFields= [], updatedFields= {"comments.13": "You will learn a lot if you read the MongoDB blog!"}},
16
  txnNumber=null,
17
  lsid=null
18
}

1
private static void exampleWithResumeToken(MongoCollection<Grade> grades) {
2
    List<Bson> pipeline = List.of(match(eq("operationType", "update")));
3
    ChangeStreamIterable<Grade> changeStream = grades.watch(pipeline);
4
    MongoChangeStreamCursor<ChangeStreamDocument<Grade>> cursor = changeStream.cursor();
5
    System.out.println("==> Going through the stream a first time & record a resumeToken");
6
    int indexOfOperationToRestartFrom = 5;
7
    int indexOfIncident = 8;
8
    int counter = 0;
9
    BsonDocument resumeToken = null;
10
    while (cursor.hasNext() && counter != indexOfIncident) {
11
        ChangeStreamDocument<Grade> event = cursor.next();
12
        if (indexOfOperationToRestartFrom == counter) {
13
            resumeToken = event.getResumeToken();
14
        }
15
        System.out.println(event);
16
        counter++;
17
    }
18
    System.out.println("==> Let's imagine something wrong happened and I need to restart my Change Stream.");
19
    System.out.println("==> Starting from resumeToken=" + resumeToken);
20
    assert resumeToken != null;
21
    grades.watch(pipeline).resumeAfter(resumeToken).forEach(printEvent());
22
}

1
==> Going through the stream a first time & record a resumeToken
2
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000012B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCCE74AA51A0486763FE0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcce74aa51a0486763fe"}}, clusterTime=Timestamp{value=6786856975532556289, seconds=1580188278, inc=1}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
3
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000022B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBA0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbba"}}, clusterTime=Timestamp{value=6786856975532556290, seconds=1580188278, inc=2}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.15": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
4
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000032B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBB0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbb"}}, clusterTime=Timestamp{value=6786856975532556291, seconds=1580188278, inc=3}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
5
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000042B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBC0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbc"}}, clusterTime=Timestamp{value=6786856975532556292, seconds=1580188278, inc=4}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
6
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000052B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBD0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbd"}}, clusterTime=Timestamp{value=6786856975532556293, seconds=1580188278, inc=5}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
7
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000062B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBE0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbe"}}, clusterTime=Timestamp{value=6786856975532556294, seconds=1580188278, inc=6}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
8
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000072B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBF0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbf"}}, clusterTime=Timestamp{value=6786856975532556295, seconds=1580188278, inc=7}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
9
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000082B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC00004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc0"}}, clusterTime=Timestamp{value=6786856975532556296, seconds=1580188278, inc=8}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
10
==> Let's imagine something wrong happened and I need to restart my Change Stream.
11
==> Starting from resumeToken={"_data": "825E2FC276000000062B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBE0004"}
12
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000072B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBF0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbbf"}}, clusterTime=Timestamp{value=6786856975532556295, seconds=1580188278, inc=7}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
13
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000082B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC00004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc0"}}, clusterTime=Timestamp{value=6786856975532556296, seconds=1580188278, inc=8}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
14
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC276000000092B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC10004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc1"}}, clusterTime=Timestamp{value=6786856975532556297, seconds=1580188278, inc=9}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
15
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000A2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC20004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc2"}}, clusterTime=Timestamp{value=6786856975532556298, seconds=1580188278, inc=10}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
16
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000B2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBC30004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbc3"}}, clusterTime=Timestamp{value=6786856975532556299, seconds=1580188278, inc=11}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"comments.14": "You will learn a lot if you read the MongoDB blog!"}}, txnNumber=null, lsid=null}
17
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000D2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC8F94B5117D894CBB90004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc8f94b5117d894cbb9"}}, clusterTime=Timestamp{value=6786856975532556301, seconds=1580188278, inc=13}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"scores.0.score": 904745.0267635228, "x": 150}}, txnNumber=null, lsid=null}
18
ChangeStreamDocument{ operationType=OperationType{value='update'}, resumeToken={"_data": "825E2FC2760000000F2B022C0100296E5A100496C525567BB74BD28BFD504F987082C046645F696400645E27BCC9F94B5117D894CBBA0004"}, namespace=sample_training.grades, destinationNamespace=null, fullDocument=null, documentKey={"_id": {"$oid": "5e27bcc9f94b5117d894cbba"}}, clusterTime=Timestamp{value=6786856975532556303, seconds=1580188278, inc=15}, updateDescription=UpdateDescription{removedFields=[], updatedFields={"scores.0.score": 2126144.0353088505, "x": 150}}, txnNumber=null, lsid=null}

1
package com.mongodb.quickstart;
2
3
import com.mongodb.ConnectionString;
4
import com.mongodb.MongoClientSettings;
5
import com.mongodb.client.*;
6
import com.mongodb.client.model.changestream.ChangeStreamDocument;
7
import com.mongodb.quickstart.models.Grade;
8
import org.bson.BsonDocument;
9
import org.bson.codecs.configuration.CodecRegistry;
10
import org.bson.codecs.pojo.PojoCodecProvider;
11
import org.bson.conversions.Bson;
12
13
import java.util.List;
14
import java.util.function.Consumer;
15
16
import static com.mongodb.client.model.Aggregates.match;
17
import static com.mongodb.client.model.Filters.eq;
18
import static com.mongodb.client.model.Filters.in;
19
import static com.mongodb.client.model.changestream.FullDocument.UPDATE_LOOKUP;
20
import static org.bson.codecs.configuration.CodecRegistries.fromProviders;
21
import static org.bson.codecs.configuration.CodecRegistries.fromRegistries;
22
23
public class ChangeStreams {
24
25
    public static void main(String[] args) {
26
        ConnectionString connectionString = new ConnectionString(System.getProperty("mongodb.uri"));
27
        CodecRegistry pojoCodecRegistry = fromProviders(PojoCodecProvider.builder().automatic(true).build());
28
        CodecRegistry codecRegistry = fromRegistries(MongoClientSettings.getDefaultCodecRegistry(), pojoCodecRegistry);
29
        MongoClientSettings clientSettings = MongoClientSettings.builder()
30
                                                                .applyConnectionString(connectionString)
31
                                                                .codecRegistry(codecRegistry)
32
                                                                .build();
33
34
        try (MongoClient mongoClient = MongoClients.create(clientSettings)) {
35
            MongoDatabase db = mongoClient.getDatabase("sample_training");
36
            MongoCollection<Grade> grades = db.getCollection("grades", Grade.class);
37
            List<Bson> pipeline;
38
39
            // Only uncomment one example at a time. Follow instructions for each individually then kill all remaining processes.
40
41
            /** => Example 1: print all the write operations.
42
             *  => Start "ChangeStreams" then "MappingPOJOs" to see some change events.
43
             */
44
            grades.watch().forEach(printEvent());
45
46
            /** => Example 2: print only insert and delete operations.
47
             *  => Start "ChangeStreams" then "MappingPOJOs" to see some change events.
48
             */
49
//             pipeline = List.of(match(in("operationType", List.of("insert", "delete"))));
50
//             grades.watch(pipeline).forEach(printEvent());
51
52
            /** => Example 3: print only updates without fullDocument.
53
             *  => Start "ChangeStreams" then "Update" to see some change events (start "Create" before if not done earlier).
54
             */
55
//             pipeline = List.of(match(eq("operationType", "update")));
56
//             grades.watch(pipeline).forEach(printEvent());
57
58
            /** => Example 4: print only updates with fullDocument.
59
             *  => Start "ChangeStreams" then "Update" to see some change events.
60
             */
61
//             pipeline = List.of(match(eq("operationType", "update")));
62
//             grades.watch(pipeline).fullDocument(UPDATE_LOOKUP).forEach(printEvent());
63
64
            /**
65
             * => Example 5: iterating using a cursor and a while loop + remembering a resumeToken then restart the Change Streams.
66
             * => Start "ChangeStreams" then "Update" to see some change events.
67
             */
68
//             exampleWithResumeToken(grades);
69
        }
70
    }
71
72
    private static void exampleWithResumeToken(MongoCollection<Grade> grades) {
73
        List<Bson> pipeline = List.of(match(eq("operationType", "update")));
74
        ChangeStreamIterable<Grade> changeStream = grades.watch(pipeline);
75
        MongoChangeStreamCursor<ChangeStreamDocument<Grade>> cursor = changeStream.cursor();
76
        System.out.println("==> Going through the stream a first time & record a resumeToken");
77
        int indexOfOperationToRestartFrom = 5;
78
        int indexOfIncident = 8;
79
        int counter = 0;
80
        BsonDocument resumeToken = null;
81
        while (cursor.hasNext() && counter != indexOfIncident) {
82
            ChangeStreamDocument<Grade> event = cursor.next();
83
            if (indexOfOperationToRestartFrom == counter) {
84
                resumeToken = event.getResumeToken();
85
            }
86
            System.out.println(event);
87
            counter++;
88
        }
89
        System.out.println("==> Let's imagine something wrong happened and I need to restart my Change Stream.");
90
        System.out.println("==> Starting from resumeToken=" + resumeToken);
91
        assert resumeToken != null;
92
        grades.watch(pipeline).resumeAfter(resumeToken).forEach(printEvent());
93
    }
94
95
    private static Consumer<ChangeStreamDocument<Grade>> printEvent() {
96
        return System.out::println;
97
    }
98
}