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

        mongoSinkConfig = MongoSinkConfig.load(config, sinkContext);
        mongoSinkConfig.validate();

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

        final MongoDatabase db = mongoClient.getDatabase(mongoSinkConfig.getDatabase());
        collection = db.getCollection(mongoSinkConfig.getCollection());

        incomingList = Lists.newArrayList();
        flushExecutor = Executors.newScheduledThreadPool(1);
        flushExecutor.scheduleAtFixedRate(() -> flush(),
                mongoSinkConfig.getBatchTimeMs(), mongoSinkConfig.getBatchTimeMs(), TimeUnit.MILLISECONDS);
    }
@Test
    public void testOpen() throws Exception {
        sink.open(map, mockSinkContext);
    }