### Question:
HistoryService { public void addAttrib2TemporalEntity(String entityId, String payload) throws ResponseException, Exception { logger.trace("replace attribute in temporal entity"); final JsonObject jsonObject = parser.parse(payload).getAsJsonObject(); String now = SerializationTools.formatter.format(Instant.now()); if (!historyDAO.entityExists(entityId)) { throw new ResponseException(ErrorType.NotFound, "You cannot create an attribute on a none existing entity"); } for (Map.Entry<String, JsonElement> entry : jsonObject.entrySet()) { logger.debug("Key = " + entry.getKey() + " Value = " + entry.getValue()); if (entry.getKey().equalsIgnoreCase(NGSIConstants.JSON_LD_ID) || entry.getKey().equalsIgnoreCase(NGSIConstants.JSON_LD_TYPE) || entry.getKey().equalsIgnoreCase(NGSIConstants.NGSI_LD_CREATED_AT) || entry.getKey().equalsIgnoreCase(NGSIConstants.NGSI_LD_MODIFIED_AT)) { continue; } String attribId = entry.getKey(); if (entry.getValue().isJsonArray()) { JsonArray valueArray = entry.getValue().getAsJsonArray(); Integer instanceCount = 0; for (JsonElement jsonElement : valueArray) { jsonElement = setCommonTemporalProperties(jsonElement, now, false); Boolean overwriteOp = (instanceCount == 0); pushAttributeToKafka(entityId, null, null, now, attribId, jsonElement.toString(), false, overwriteOp); instanceCount++; } } } logger.trace("attribute replaced in temporalentity " + entityId); } HistoryService(ProducerChannel producerChannels); URI createTemporalEntityFromEntity(String payload); URI createTemporalEntityFromBinding(String payload); void delete(String entityId, String attributeId, String instanceId, List<Object> linkHeaders); void addAttrib2TemporalEntity(String entityId, String payload); void modifyAttribInstanceTemporalEntity(String entityId, String payload, String attribId, String instanceId,
			List<Object> linkHeaders); @KafkaListener(topics = "${entity.create.topic}", groupId = "historyManagerCreate") void handleEntityCreate(@Payload byte[] message, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key); @KafkaListener(topics = "${entity.append.topic}", groupId = "historyManagerAppend") void handleEntityAppend(@Payload byte[] message, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key); @KafkaListener(topics = "${entity.update.topic}", groupId = "historyManagerUpdate") void handleEntityUpdate(@Payload byte[] message, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key); @KafkaListener(topics = "${entity.delete.topic}", groupId = "historyManagerDelete") void handleEntityDelete(@Payload byte[] message, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key);  }### Answer:
@Test public void addAttrib2TemporalEntityTest() { try { Mockito.doReturn(true).when(historyDAO).entityExists(any()); historyService.addAttrib2TemporalEntity("urn:ngsi-ld:testunit:151", temporalPayload); verify(kafkaOperations, times(3)).pushToKafka(any(),any(),any()); } catch (Exception e) { Assert.fail(); e.printStackTrace(); } }