### Question:
DaprStateAsyncProvider { Mono<Void> apply(String actorType, ActorId actorId, ActorStateChange... stateChanges) { if ((stateChanges == null) || stateChanges.length == 0) { return Mono.empty(); } int count = 0; byte[] payload = null; try (ByteArrayOutputStream writer = new ByteArrayOutputStream()) { JsonGenerator generator = JSON_FACTORY.createGenerator(writer); generator.writeStartArray(); for (ActorStateChange stateChange : stateChanges) { if ((stateChange == null) || (stateChange.getChangeKind() == null)) { continue; } String operationName = stateChange.getChangeKind().getDaprStateChangeOperation(); if ((operationName == null) || (operationName.length() == 0)) { continue; } count++; generator.writeStartObject(); generator.writeStringField("operation", operationName); generator.writeObjectFieldStart("request"); generator.writeStringField("key", stateChange.getStateName()); if ((stateChange.getChangeKind() == ActorStateChangeKind.UPDATE) || (stateChange.getChangeKind() == ActorStateChangeKind.ADD)) { byte[] data = this.stateSerializer.serialize(stateChange.getValue()); if (data != null) { if (this.isStateSerializerDefault) { generator.writeFieldName("value"); generator.writeRawValue(new String(data, CHARSET)); } else { generator.writeBinaryField("value", data); } } } generator.writeEndObject(); generator.writeEndObject(); } generator.writeEndArray(); generator.close(); writer.flush(); payload = writer.toByteArray(); } catch (IOException e) { e.printStackTrace(); return Mono.error(e); } if (count == 0) { Mono.empty(); } return this.daprClient.saveActorStateTransactionally(actorType, actorId.toString(), payload); } DaprStateAsyncProvider(DaprClient daprClient, DaprObjectSerializer stateSerializer);   }### Answer:
@Test public void happyCaseApply() { DaprClient daprClient = mock(DaprClient.class); when(daprClient .saveActorStateTransactionally( eq("MyActor"), eq("123"), argThat(s -> { try { JsonNode node = OBJECT_MAPPER.readTree(s); if (node == null) { return false; } if (node.size() != 3) { return false; } boolean foundInsertName = false; boolean foundUpdateZipcode = false; boolean foundDeleteFlag = false; for (JsonNode operation : node) { if (operation.get("operation") == null) { return false; } if (operation.get("request") == null) { return false; } String opName = operation.get("operation").asText(); String key = operation.get("request").get("key").asText(); JsonNode valueNode = operation.get("request").get("value"); String value = (valueNode == null) ? null : valueNode.asText(); foundInsertName |= "upsert".equals(opName) && "name".equals(key) && "Jon Doe".equals(value); foundUpdateZipcode |= "upsert".equals(opName) && "zipcode".equals(key) && "98011".equals(value); foundDeleteFlag |= "delete".equals(opName) && "flag".equals(key) && (value == null); } return foundInsertName && foundUpdateZipcode && foundDeleteFlag; } catch (IOException e) { e.printStackTrace(); return false; } }))) .thenReturn(Mono.empty()); DaprStateAsyncProvider provider = new DaprStateAsyncProvider(daprClient, SERIALIZER); provider.apply("MyActor", new ActorId("123"), createInsertChange("name", "Jon Doe"), createUpdateChange("zipcode", 98011), createDeleteChange("flag")) .block(); verify(daprClient).saveActorStateTransactionally(eq("MyActor"), eq("123"), any()); }