@Override
    public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
        log.info("Open MongoDB Source");

        mongoSourceConfig = MongoSourceConfig.load(config, sourceContext);
        mongoSourceConfig.validate();

        if (clientProvider != null) {
            mongoClient = clientProvider.get();
        } else {
            mongoClient = MongoClients.create(mongoSourceConfig.getMongoUri());
        }

        String mongoDatabase = mongoSourceConfig.getDatabase();
        if (StringUtils.isEmpty(mongoDatabase)) {
            // Watch all databases
            log.info("Watch all databases");
            stream = mongoClient.watch();

        } else {
            final MongoDatabase db = mongoClient.getDatabase(mongoDatabase);
            String mongoCollection = mongoSourceConfig.getCollection();
            if (StringUtils.isEmpty(mongoCollection)) {
                // Watch all collections in a database
                log.info("Watch db: {}", db.getName());
                stream = db.watch();

            } else {
                // Watch a collection
                final MongoCollection<Document> collection = db.getCollection(mongoCollection);
                log.info("Watch collection: {}.{}", db.getName(), mongoCollection);
                stream = collection.watch();
            }
        }

        stream.batchSize(mongoSourceConfig.getBatchSize())
                .fullDocument(FullDocument.UPDATE_LOOKUP);

        if (mongoSourceConfig.getSyncType() == SyncType.FULL_SYNC) {
            // sync currently existing messages
            // startAtOperationTime is the starting point for the change stream
            // setting startAtOperationTime to 0 means the start point is the earliest
            // see https://www.mongodb.com/docs/v4.2/reference/method/db.collection.watch/ for more information
            stream.startAtOperationTime(new BsonTimestamp(0L));
        }

        stream.subscribe(new Subscriber<ChangeStreamDocument<Document>>() {
            private ObjectMapper mapper = new ObjectMapper();
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                this.subscription.request(Integer.MAX_VALUE);
            }

            @Override
            public void onNext(ChangeStreamDocument<Document> doc) {
                try {
                    log.info("New change doc: {}", doc);

                    BsonDocument documentKey = doc.getDocumentKey();
                    if (documentKey == null) {
                        log.warn("The document key is null");
                        return;
                    }

                    // Build a record with the essential information
                    final Map<String, Object> recordValue = new HashMap<>();
                    recordValue.put("fullDocument", doc.getFullDocument());
                    recordValue.put("ns", doc.getNamespace());
                    recordValue.put("operation", doc.getOperationType());

                    consume(new DocRecord(
                            Optional.of(documentKey.toJson()),
                            mapper.writeValueAsString(recordValue).getBytes(StandardCharsets.UTF_8)));

                } catch (JsonProcessingException e) {
                    log.error("Processing doc from mongo", e);
                }
            }

            @Override
            public void onError(Throwable error) {
                log.error("Subscriber error", error);
            }

            @Override
            public void onComplete() {
                log.info("Subscriber complete");
            }
        });

    }
@Test
    public void testOpen() throws Exception {
        source.open(map, mockSourceContext);
    }