@Test
public void testCreateRetryWithError() {
    Repository<Entity> mockRepo = Mockito.spy(repository);
    Map<String, Object> studentBody = buildTestStudentEntity();
    Map<String, Object> studentMetaData = new HashMap<String, Object>();
    int noOfRetries = 5;

    Mockito.doThrow(new MongoException("Test Exception"))
            .when(((MongoEntityRepository) mockRepo))
            .internalCreate("student", null, studentBody, studentMetaData, "student");
    Mockito.doCallRealMethod()
            .when(mockRepo)
            .createWithRetries("student", null, studentBody, studentMetaData, "student",
                    noOfRetries);

    try {
        mockRepo.createWithRetries("student", null, studentBody, studentMetaData, "student",
                noOfRetries);
    } catch (MongoException ex) {
        assertEquals(ex.getMessage(), "Test Exception");
    }

    Mockito.verify((MongoEntityRepository) mockRepo, Mockito.times(noOfRetries))
            .internalCreate("student", null, studentBody, studentMetaData, "student");
}
 

private MongoClient createClient(AbstractConfig config, MongoClientOptions options) {
    String host = config.getString(MONGO_HOST);
    int port = config.getInt(MONGO_PORT);

    try {
        MongoClientOptions actualOptions;
        if (options != null) {
            actualOptions = options;
        } else {
            actualOptions = new MongoClientOptions.Builder().build();
        }
        ServerAddress server = new ServerAddress(host, port);
        if (credentials != null) {
            return new MongoClient(server, credentials, actualOptions);
        } else {
            return new MongoClient(server, actualOptions);
        }
    } catch (MongoException ex) {
        log.error("Failed to create MongoDB client to {}:{} with credentials {}", host, port,
                credentials, ex);
        throw new ConnectException("MongoDb client cannot be created.", ex);
    }
}
 

@SuppressWarnings( "rawtypes" )
public cfData execute(cfSession _session, cfArgStructData argStruct ) throws cfmRunTimeException {
	MongoDatabase db	= getMongoDatabase( _session, argStruct );
	cfData	cmdData	= getNamedParam(argStruct, "cmd", null );
	if ( cmdData == null )
		throwException(_session, "please specify the cmd parameter");
	
	try{
		Document cmr;
		
		if ( cmdData.getDataType() == cfData.CFSTRUCTDATA )
			cmr	= db.runCommand( getDocument( (cfStructData) cmdData ) );
		else
			cmr	= db.runCommand( new Document( cmdData.getString(), true ) );
		
		return tagUtils.convertToCfData((Map)cmr);
	} catch (MongoException me){
		throwException(_session, me.getMessage());
		return null;
	}
}
 

boolean update(String collectionName, Entity entity, List<Entity> failed, AbstractMessageReport report,
        ReportStats reportStats, Source nrSource) {
    boolean res = false;

    try {
        res = entityRepository.updateWithRetries(collectionName, entity, totalRetries);
        if (!res) {
            failed.add(entity);
        }
    } catch (MongoException e) {
        NestedRuntimeException wrapper = new NestedRuntimeException("Mongo Exception", e) {
            private static final long serialVersionUID = 1L;
        };
        reportWarnings(wrapper.getMostSpecificCause().getMessage(), collectionName,
                ((SimpleEntity) entity).getSourceFile(), report, reportStats, nrSource);
    }

    return res;
}
 

@Override
protected void doProcess(JCas jCas) throws AnalysisEngineProcessException {

  String documentId = ConsumerUtils.getExternalId(getDocumentAnnotation(jCas), contentHashAsId);

  // Delete any existing content in the database
  deleteAllContent(documentId);

  // Save
  try {
    saveEvents(documentId, jCas, textClass);
  } catch (MongoException | BsonSerializationException e) {
    getMonitor()
        .error(
            "Unable to persist relations to database - document {} will contain no relations",
            getDocumentAnnotation(jCas).getSourceUri(),
            e);
  }
}
 

private void init(MongoDbAccessor mongo) {
    LOG.info(">>> init");

    try {
        final MongoCollection<Document> profileCollection = getProfileCollection(mongo);

        IndexOptions indexOptions = new IndexOptions();
        indexOptions.background(true);
        LOG.info("Create index {ts:-1, lbl:1} in the background if it does not yet exists");
        profileCollection.createIndex(new BasicDBObject("ts",-1).append("lbl", 1), indexOptions);
        LOG.info("Create index {adr:1, db:1, ts:-1} in the background if it does not yet exists");
        profileCollection.createIndex(new BasicDBObject("adr",1).append("db",1).append("ts", -1), indexOptions);
        ApplicationStatusDto.addWebLog("ProfilingWriter is successfully connected to its collector database.");
    } catch (MongoException e) {
        LOG.error("Exception while connecting to: {}", serverDto.getHosts(), e);
        ApplicationStatusDto.addWebLog("ProfilingWriter could not connect to its collector database.");
    }
    
    LOG.info("<<< init");
}
 

@Test
public void testDeSerializeStatementToDocument() throws RyaDAOException, MongoException, IOException {
    RyaStatement statement = storageStrategy.deserializeDocument(TEST_DOC);
    /*
     * Since RyaStatement creates a timestamp using JVM time if the timestamp is null, we want to re-null it
     * for this test.  Timestamp is created at insert time by the Server, this test
     * can be found in the RyaDAO.
     */
    statement.setTimestamp(null);
    assertEquals(testStatement, statement);

    statement = storageStrategy.deserializeDocument(TEST_DOC_2);
    /*
     * Since RyaStatement creates a timestamp using JVM time if the timestamp is null, we want to re-null it
     * for this test.  Timestamp is created at insert time by the Server, this test
     * can be found in the RyaDAO.
     */
    statement.setTimestamp(null);
    assertEquals(testStatement2, statement);
}
 

@Test
public void testAdd() throws RyaDAOException, MongoException, IOException {
    final MongoDBRyaDAO dao = new MongoDBRyaDAO();
    try {
        dao.setConf(conf);
        dao.init();

        final RyaStatementBuilder builder = new RyaStatementBuilder();
        builder.setPredicate(new RyaIRI("http://temp.com"));
        builder.setSubject(new RyaIRI("http://subject.com"));
        builder.setObject(new RyaIRI("http://object.com"));

        final MongoDatabase db = conf.getMongoClient().getDatabase(conf.get(MongoDBRdfConfiguration.MONGO_DB_NAME));
        final MongoCollection<Document> coll = db.getCollection(conf.getTriplesCollectionName());

        dao.add(builder.build());

        assertEquals(coll.countDocuments(), 1);
    }  finally {
        dao.destroy();
    }
}
 

@Test
public void testDelete() throws RyaDAOException, MongoException, IOException {
    final MongoDBRyaDAO dao = new MongoDBRyaDAO();
    try {
        dao.setConf(conf);
        dao.init();

        final RyaStatementBuilder builder = new RyaStatementBuilder();
        builder.setPredicate(new RyaIRI("http://temp.com"));
        builder.setSubject(new RyaIRI("http://subject.com"));
        builder.setObject(new RyaIRI("http://object.com"));
        final RyaStatement statement = builder.build();

        final MongoDatabase db = conf.getMongoClient().getDatabase(conf.get(MongoDBRdfConfiguration.MONGO_DB_NAME));
        final MongoCollection<Document> coll = db.getCollection(conf.getTriplesCollectionName());

        dao.add(statement);
        assertEquals(coll.countDocuments(), 1);

        dao.delete(statement, conf);
        assertEquals(coll.countDocuments(), 0);
    } finally {
        dao.destroy();
    }
}
 

@Test
public void testDelete() throws RyaDAOException, MongoException, IOException {
    final MongoDBRyaDAO dao = new MongoDBRyaDAO();
    try {
        dao.setConf(conf);
        dao.init();

        final RyaStatementBuilder builder = new RyaStatementBuilder();
        builder.setPredicate(new RyaIRI("http://temp.com"));
        builder.setSubject(new RyaIRI("http://subject.com"));
        builder.setObject(new RyaIRI("http://object.com"));
        builder.setColumnVisibility(new DocumentVisibility("C").flatten());
        final RyaStatement statement = builder.build();
        final MongoDatabase db = conf.getMongoClient().getDatabase(conf.get(MongoDBRdfConfiguration.MONGO_DB_NAME));
        final MongoCollection<Document> coll = db.getCollection(conf.getTriplesCollectionName());

        dao.add(statement);
        assertEquals(1, coll.countDocuments());

        dao.delete(statement, conf);
        assertEquals(0, coll.countDocuments());
    } finally {
        dao.destroy();
    }
}
 

@SuppressWarnings("boxing")
public static String ensureIndexes(MongoIndex index, DB dbConn, int indexOrder) {
    DBObject options = new BasicDBObject();
    options.put("name", "idx_" + indexOrder);
    options.put("unique", index.isUnique());
    options.put("sparse", index.isSparse());
    options.put("ns", dbConn.getCollection(index.getCollection()).getFullName());

    try {
        dbConn.getCollection(index.getCollection()).createIndex(new BasicDBObject(index.getKeys()), options);
        return null;
    } catch (MongoException e) {
        LOG.error("Failed to ensure index:{}", e.getMessage());
        return "Failed to ensure index:" + e.getMessage();
    }
}
 

@Override
public Optional<Type> get(final RyaIRI typeId) throws TypeStorageException {
    requireNonNull(typeId);

    try {
        final Document document = mongo.getDatabase(ryaInstanceName)
            .getCollection(COLLECTION_NAME)
            .find( makeIdFilter(typeId) )
            .first();

        return document == null ?
                Optional.empty() :
                Optional.of( TYPE_CONVERTER.fromDocument(document) );

    } catch(final MongoException | DocumentConverterException e) {
        throw new TypeStorageException("Could not get the Type with ID '" + typeId.getData() + "'.", e);
    }
}
 

@Override
public int _countByCriteria(Table table) {
	int count = 0;
	Map<String, Object> coditon = new HashMap<String, Object>();
	try {
		QueryCriteria queryCriteria = table.getQueryCriteria();
		DB db = database.getDB();
		Jongo jongo = new Jongo(db);
		for(Criteria criteria:queryCriteria.getOredCriteria()){
			for(Criterion criterion:criteria.getAllCriteria()){
				coditon = buildCriteria(criterion, coditon);
			}
		}
		long size = jongo.getCollection(queryCriteria.getTable()).count(JsonUtils.objToJson(coditon));
		count = (int) size;
	} catch (MongoException e) {
		LOG.error("mongo find error", e);
	}
	return count;
}
 

public ProfilingWriter(BlockingQueue<ProfilingEntry> jobQueue) {
    this.jobQueue = jobQueue;
    serverDto = ConfigReader.getCollectorServer();
    runningSince = new Date();

    final MongoDbAccessor mongo = getMongoDbAccessor();
    try {
        final MongoCollection<Document> profileCollection = getProfileCollection(mongo);

        IndexOptions indexOptions = new IndexOptions();
        indexOptions.background(true);
        LOG.info("Create index {ts:-1, lbl:1} in the background if it does not yet exists");
        profileCollection.createIndex(new BasicDBObject("ts",-1).append("lbl", 1), indexOptions);
        LOG.info("Create index {adr:1, db:1, ts:-1} in the background if it does not yet exists");
        profileCollection.createIndex(new BasicDBObject("adr",1).append("db",1).append("ts", -1), indexOptions);

        LOG.info("ProfilingWriter is ready at {}", serverDto.getHosts());

    } catch (MongoException e) {
        LOG.error("Exception while connecting to: {}", serverDto.getHosts(), e);
    }
}
 

@Override
public int _countByCriteria(Table table) {
	int count = 0;
	Map<String, Object> coditon = new HashMap<String, Object>();
	try {
		QueryCriteria queryCriteria = table.getQueryCriteria();
		com.mongodb.client.MongoDatabase db = database.getMongoDB();
		for(Criteria criteria:queryCriteria.getOredCriteria()){
			for(Criterion criterion:criteria.getAllCriteria()){
				coditon = buildCriteria(criterion, coditon);
			}
		}
		long size = db.getCollection(queryCriteria.getTable()).count(Document.parse((JSON.serialize(coditon))));
		LOG.debug("countByCriteria->collection:"+queryCriteria.getTable()+",script:"+JSON.serialize(coditon));
		count = (int) size;
	} catch (MongoException e) {
		LOG.error("mongo find error", e);
	}
	LOG.debug("_countByCriteria->result:"+count);
	return count;
}
 

public cfData execute(cfSession _session, cfArgStructData argStruct ) throws cfmRunTimeException {
	MongoDatabase	db	= getMongoDatabase( _session, argStruct );
	
	String collection	= getNamedStringParam(argStruct, "collection", null);
	if ( collection == null )
		throwException(_session, "please specify a collection");
	
	String name	= getNamedStringParam(argStruct, "name", null);
	if ( name == null )
		throwException(_session, "please specify a name");
	
	try{
		
		db
			.getCollection( collection )
			.renameCollection( new MongoNamespace( db.getName(), name ), new RenameCollectionOptions().dropTarget( getNamedBooleanParam(argStruct, "droptarget", false ) ) );

		return cfBooleanData.TRUE;
		
	} catch (MongoException me){
		throwException(_session, me.getMessage());
		return null;
	}
}
 

public cfData execute( cfSession _session, cfArgStructData argStruct ) throws cfmRunTimeException {
	MongoClient client = getMongoClient( _session, argStruct );

	try {
		cfArrayData arr = cfArrayData.createArray( 1 );

		client.listDatabaseNames().forEach( new Block<String>() {

			@Override
			public void apply( final String st ) {
				try {
					arr.addElement( new cfStringData( st ) );
				} catch ( cfmRunTimeException e ) {}
			}
		} );

		return arr;
	} catch ( MongoException me ) {
		throwException( _session, me.getMessage() );
		return null;
	}
}
 

@Override
public int _updateByCriteria(Table table) {
	Map<String, Object> coditon = new HashMap<String, Object>();
	try {
        QueryCriteria queryCriteria = table.getQueryCriteria();
        com.mongodb.client.MongoDatabase db = database.getMongoDB();
		for(Criteria criteria:queryCriteria.getOredCriteria()){
			for(Criterion criterion:criteria.getAllCriteria()){
				coditon = buildCriteria(criterion, coditon);
			}
		}
		Map<String, Object> vaule = new HashMap<String, Object>();
		vaule.put("$set", table.getParams());
		db.getCollection(queryCriteria.getTable()).updateMany(Document.parse((JSON.serialize(coditon))), Document.parse(JSON.serialize(vaule)));
		LOG.debug("updateByCriteria->collection:"+table.getTableName()+",value:"+JSON.serialize(vaule)+",condition:"+JSON.serialize(coditon));
	} catch (MongoException e) {
		LOG.error("mongo update error", e);
	}
	return 1;
}
 

public cfData execute(cfSession _session, cfArgStructData argStruct ) throws cfmRunTimeException {
	MongoDatabase	db	= getMongoDatabase( _session, argStruct );
	
	String collection	= getNamedStringParam(argStruct, "collection", null);
	if ( collection == null )
		throwException(_session, "please specify a collection");
	
	String index	= getNamedStringParam(argStruct, "index", null );
	
	try{
		if ( index != null )
			db.getCollection(collection).dropIndex(index);
		else
			db.getCollection(collection).dropIndexes();

		return cfBooleanData.TRUE;
	} catch (MongoException me){
		throwException(_session, me.getMessage());
		return null;
	}
}
 

@Override
public AttachmentMetadata getMetadata(AttachmentId attachmentId) throws IOException {
  GridFSFile metadataFile = getMetadataGridFSFile(attachmentId);

  if (metadataFile == null)
    return null;

  try (GridFSDownloadStream metadataStream = metadataGrid
      .openDownloadStream(metadataFile.getObjectId())) {

    if (metadataStream == null) {
      return null;
    }

    AttachmentProto.AttachmentMetadata protoMetadata = AttachmentProto.AttachmentMetadata
        .parseFrom(metadataStream);

    return new AttachmentMetadataProtoImpl(protoMetadata);

  } catch (MongoException e) {
    throw new IOException(e);
  }

}
 

@Override
public T decode(BsonReader reader, DecoderContext decoderContext) {
  RawBsonDocument doc = codec.decode(reader, decoderContext);
  try {
    return mapper
        .readerWithView(DatabaseOnly.class)
        .forType(this.clazz)
        .readValue(doc.getByteBuffer().array());
  } catch (IOException e) {
    throw new MongoException(e.getMessage());
  }
}
 

@Override
public void encode(BsonWriter writer, T value, EncoderContext encoderContext) {
  try {
    byte[] data = mapper.writerWithView(DatabaseOnly.class).writeValueAsBytes(value);
    codec.encode(writer, new RawBsonDocument(data), encoderContext);
  } catch (JsonProcessingException e) {
    throw new MongoException(e.getMessage());
  }
}
 

public void persist(PersistentObject object) {
  try {
    MongoCollection<PersistentObject> coll = this.getCollection(object);

    coll.replaceOne(
        new Document(FIELD_ID, object.getId()), object, new UpdateOptions().upsert(true));
  } catch (MongoException e) {
    LOGGER.error("Error while saving into database: {}", e.getMessage());
  }
}
 

private void checkRetriableException(final MongoSinkTopicConfig config, final MongoException e) {
  if (getRemainingRetriesForTopic(config.getTopic()).decrementAndGet() <= 0) {
    throw new DataException("Failed to write mongodb documents despite retrying", e);
  }
  Integer deferRetryMs = config.getInt(RETRIES_DEFER_TIMEOUT_CONFIG);
  LOGGER.debug("Deferring retry operation for {}ms", deferRetryMs);
  context.timeout(deferRetryMs);
  throw new RetriableException(e.getMessage(), e);
}
 

@Override
public void submitTask(RecoverableAsyncTask task) {
    
    // Get the recovery 
    RecoveryRecord record = task.getRecoveryRecord();
    
    // If we intend to process, write the recovery record
    if(isProcessingLocally()){
        record.markAsProcessing(this.signature);
    }
    
    try{
        // needs to be journaled 
        this.recoveryColl.insert(record.toDBObject());           
    }
    catch(MongoException ex){
        
        // If we failed to write at all, this is a complete
        // failure, we need to flow this out to the caller
        // or try to do the post synchronously
        throw ServiceException.wrap(ex, AsyncError.CANNOT_WRITE_RECOVERY_RECORD);
    }

    // If this is a processor instance, post it to the executor
    if(isProcessingLocally()){
        this.executor.execute(task);
    }
}
 

@Async
@Override
public void saveData(ZipInputStream zipInputStream, ZipFile zipFile) throws MongoException, IOException {
    ZipEntry zipEntry;
    while ((zipEntry = zipInputStream.getNextEntry()) != null) {
        if (zipEntry.toString().endsWith("txt")) {
            BufferedReader br = new BufferedReader(
                    new InputStreamReader(zipFile.getInputStream(zipEntry)));
            saveHistoryDataFromTxt(br);
            br.close();

        }
    }
}
 

private void processSinkRecords(MongoCollection<BsonDocument> collection, List<SinkRecord> batch) {
    String collectionName = collection.getNamespace().getCollectionName();
    List<? extends WriteModel<BsonDocument>> docsToWrite =
            sinkConfig.isUsingCdcHandler(collectionName)
                    ? buildWriteModelCDC(batch,collectionName)
                    : buildWriteModel(batch,collectionName);
    try {
        if (!docsToWrite.isEmpty()) {
            LOGGER.debug("bulk writing {} document(s) into collection [{}]",
                    docsToWrite.size(), collection.getNamespace().getFullName());
            BulkWriteResult result = collection.bulkWrite(
                    docsToWrite, BULK_WRITE_OPTIONS);
            LOGGER.debug("mongodb bulk write result: " + result.toString());
        }
    } catch (MongoException mexc) {
        if (mexc instanceof BulkWriteException) {
            BulkWriteException bwe = (BulkWriteException) mexc;
            LOGGER.error("mongodb bulk write (partially) failed", bwe);
            LOGGER.error(bwe.getWriteResult().toString());
            LOGGER.error(bwe.getWriteErrors().toString());
            LOGGER.error(bwe.getWriteConcernError().toString());
        } else {
            LOGGER.error("error on mongodb operation", mexc);
            LOGGER.error("writing {} document(s) into collection [{}] failed -> remaining retries ({})",
                    docsToWrite.size(), collection.getNamespace().getFullName() ,remainingRetries);
        }
        if (remainingRetries-- <= 0) {
            throw new ConnectException("failed to write mongodb documents"
                    + " despite retrying -> GIVING UP! :( :( :(", mexc);
        }
        LOGGER.debug("deferring retry operation for {}ms", deferRetryMs);
        context.timeout(deferRetryMs);
        throw new RetriableException(mexc.getMessage(), mexc);
    }
}
 

@Test(expected = MongoException.class)
public void testExceptionForIndexMismatch() {
    getDs().find(IndexedEntity.class).iterator(new FindOptions()
                                                   .limit(1)
                                                   .min(new Document("doesNotExist", 1)))
           .next();
}
 

/**
 * Ensures current exception has been generated due to a duplicate (primary) key.
 * Differentiates between Fongo and Mongo exceptions since the behaviour under these databases
 * is different.
 */
public static void assertDuplicateKeyException(Throwable exception) {
  Preconditions.checkNotNull(exception, "exception");

  // unwrap, if necessary
  exception = exception instanceof MongoException ? exception : exception.getCause();

  // fongo throws directly DuplicateKeyException
  if (exception instanceof DuplicateKeyException) return;

  // MongoDB throws custom exception
  if (exception instanceof MongoCommandException) {
    String codeName = ((MongoCommandException) exception).getResponse().get("codeName").asString().getValue();
    int errorCode = ((MongoCommandException) exception).getErrorCode();

    check(codeName).is("DuplicateKey");
    check(errorCode).is(11000); // code 11000 stands for DuplicateKeyException

    // all good here (can return)
    return;
  }

  // for bulk writes as well
  if (exception instanceof MongoBulkWriteException) {
    List<BulkWriteError> errors = ((MongoBulkWriteException) exception).getWriteErrors();
    check(errors).hasSize(1);
    check(errors.get(0).getCode()).is(11000);
    check(errors.get(0).getMessage()).contains("duplicate key");
    return;
  }

  // if we got here means there is a problem (no duplicate key exception)
  fail("Should get duplicate key exception after " + exception);
}
 

/**
 * Checks if the given error is caused due to duplicate keys.
 *
 * @param error The error to check.
 * @return {@code true} if the given error is caused by duplicate keys.
 * @throws NullPointerException if the error is {@code null}.
 */
public static boolean isDuplicateKeyError(final Throwable error) {
    Objects.requireNonNull(error);

    if (error instanceof MongoException) {
        final MongoException mongoException = (MongoException) error;
        return ErrorCategory.fromErrorCode(mongoException.getCode()) == ErrorCategory.DUPLICATE_KEY;
    }
    return false;
}