BulkWriteResultAckFlow { static BulkWriteResultAckFlow of(final ActorRef updaterShard) { return new BulkWriteResultAckFlow(updaterShard); } private  BulkWriteResultAckFlow(final ActorRef updaterShard);   }
@Test public void allSuccess() { final List<AbstractWriteModel> writeModels = generate5WriteModels(); final BulkWriteResult result = BulkWriteResult.acknowledged(0, 3, 1, 1, List.of(new BulkWriteUpsert(0, new BsonString("upsert 0")), new BulkWriteUpsert(4, new BsonString("upsert 4"))) ); final WriteResultAndErrors resultAndErrors = WriteResultAndErrors.success(writeModels, result); final String message = runBulkWriteResultAckFlowAndGetFirstLogEntry(resultAndErrors); actorSystem.log().info(message); assertThat(message).contains("Acknowledged: Success"); }
@Test public void partialSuccess() { final List<AbstractWriteModel> writeModels = generate5WriteModels(); final BulkWriteResult result = BulkWriteResult.acknowledged(1, 2, 1, 2, List.of()); final List<BulkWriteError> updateFailure = List.of( new BulkWriteError(11000, "E11000 duplicate key error", new BsonDocument(), 3), new BulkWriteError(50, "E50 operation timed out", new BsonDocument(), 4) ); final WriteResultAndErrors resultAndErrors = WriteResultAndErrors.failure(writeModels, new MongoBulkWriteException(result, updateFailure, null, new ServerAddress())); final String message = runBulkWriteResultAckFlowAndGetFirstLogEntry(resultAndErrors); actorSystem.log().info(message); for (int i = 3; i < 5; ++i) { assertThat(expectUpdateThingResponse(writeModels.get(i).getMetadata().getThingId())) .describedAs("response is failure") .returns(false, UpdateThingResponse::isSuccess); } assertThat(message).contains("Acknowledged: PartialSuccess"); }
@Test public void errorIndexOutOfBoundError() { final List<AbstractWriteModel> writeModels = generate5WriteModels(); final BulkWriteResult result = BulkWriteResult.acknowledged(1, 2, 1, 2, List.of()); final List<BulkWriteError> updateFailure = List.of( new BulkWriteError(11000, "E11000 duplicate key error", new BsonDocument(), 0), new BulkWriteError(50, "E50 operation timed out", new BsonDocument(), 5) ); final WriteResultAndErrors resultAndErrors = WriteResultAndErrors.failure(writeModels, new MongoBulkWriteException(result, updateFailure, null, new ServerAddress())); final String message = runBulkWriteResultAckFlowAndGetFirstLogEntry(resultAndErrors); actorSystem.log().info(message); for (final AbstractWriteModel writeModel : writeModels) { final UpdateThingResponse response = expectUpdateThingResponse(writeModel.getMetadata().getThingId()); assertThat(response).describedAs("response is failure").returns(false, UpdateThingResponse::isSuccess); } assertThat(message).contains("ConsistencyError[indexOutOfBound]"); }