@Test public void testOneCreateDataSetMessage() { final NotificationSender sender = new NotificationSender(); final Referenceable queue1 = createRef(TYPE_NIFI_QUEUE, "queue1@test"); final List<HookNotification.HookNotificationMessage> messages = Collections.singletonList( new HookNotification.EntityCreateRequest(NIFI_USER, queue1)); final Notifier notifier = new Notifier(); sender.send(messages, notifier); assertCreateMessage(notifier, 0, queue1); }
void send(final List<HookNotification.HookNotificationMessage> messages, final Consumer<List<HookNotification.HookNotificationMessage>> notifier) { final Metrics metrics = new Metrics(); try { metrics.totalMessages = messages.size(); final Map<Boolean, List<HookNotification.HookNotificationMessage>> createAndOthers = messages.stream().collect(groupingBy(msg -> ENTITY_CREATE.equals(msg.getType()))); final List<HookNotification.HookNotificationMessage> creates = safeGet(createAndOthers, true); metrics.createMessages = creates.size(); final Map<Boolean, List<Referenceable>> newFlowPathsAndOtherEntities = creates.stream() .flatMap(msg -> ((HookNotification.EntityCreateRequest) msg).getEntities().stream()) .collect(groupingBy(ref -> TYPE_NIFI_FLOW_PATH.equals(ref.typeName))); final List<Referenceable> newEntitiesExceptFlowPaths = safeGet(newFlowPathsAndOtherEntities, false) .stream().filter(distinctReferenceable()).collect(Collectors.toList()); final Collection<Referenceable> newFlowPaths = safeGet(newFlowPathsAndOtherEntities, true).stream() .collect(toMap(ref -> ref.get(ATTR_QUALIFIED_NAME), ref -> ref, (r1, r2) -> { mergeRefs(r1, r2); return r1; })).values(); metrics.uniqueFlowPathCreates = newFlowPaths.size(); metrics.uniqueOtherCreates = newEntitiesExceptFlowPaths.size(); List<Referenceable> newEntities = new ArrayList<>(); newEntities.addAll(newEntitiesExceptFlowPaths); newEntities.addAll(newFlowPaths); if (!newEntities.isEmpty()) { notifier.accept(Collections.singletonList(new HookNotification.EntityCreateRequest(NIFI_USER, newEntities))); } final Map<Boolean, List<HookNotification.HookNotificationMessage>> partialNiFiFlowPathUpdateAndOthers = safeGet(createAndOthers, false).stream().collect(groupingBy(msg -> ENTITY_PARTIAL_UPDATE.equals(msg.getType()) && TYPE_NIFI_FLOW_PATH.equals(((HookNotification.EntityPartialUpdateRequest)msg).getTypeName()) && ATTR_QUALIFIED_NAME.equals(((HookNotification.EntityPartialUpdateRequest)msg).getAttribute()) )); final List<HookNotification.HookNotificationMessage> partialNiFiFlowPathUpdates = safeGet(partialNiFiFlowPathUpdateAndOthers, true); final List<HookNotification.HookNotificationMessage> otherMessages = safeGet(partialNiFiFlowPathUpdateAndOthers, false); metrics.partialNiFiFlowPathUpdates = partialNiFiFlowPathUpdates.size(); metrics.otherMessages = otherMessages.size(); final List<HookNotification.HookNotificationMessage> deduplicatedMessages = partialNiFiFlowPathUpdates.stream().map(msg -> (HookNotification.EntityPartialUpdateRequest) msg) .collect(groupingBy(HookNotification.EntityPartialUpdateRequest::getAttributeValue)).entrySet().stream() .map(entry -> { final String flowPathQualifiedName = entry.getKey(); final Map<String, Referenceable> distinctInputs; final Map<String, Referenceable> distinctOutputs; final String flowPathGuid; try { metrics.flowPathSearched++; final AtlasEntity.AtlasEntityWithExtInfo flowPathExt = atlasClient.searchEntityDef(new AtlasObjectId(TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName)); final AtlasEntity flowPathEntity = flowPathExt.getEntity(); flowPathGuid = flowPathEntity.getGuid(); distinctInputs = toReferenceables(flowPathEntity.getAttribute(ATTR_INPUTS), metrics); distinctOutputs = toReferenceables(flowPathEntity.getAttribute(ATTR_OUTPUTS), metrics); } catch (AtlasServiceException e) { if (ClientResponse.Status.NOT_FOUND.equals(e.getStatus())) { logger.debug("nifi_flow_path was not found for qualifiedName {}", flowPathQualifiedName); } else { logger.warn("Failed to retrieve nifi_flow_path with qualifiedName {} due to {}", flowPathQualifiedName, e, e); } return null; } for (HookNotification.EntityPartialUpdateRequest msg : entry.getValue()) { fromReferenceable(msg.getEntity().get(ATTR_INPUTS), metrics) .entrySet().stream().filter(ref -> !distinctInputs.containsKey(ref.getKey())) .forEach(ref -> distinctInputs.put(ref.getKey(), ref.getValue())); fromReferenceable(msg.getEntity().get(ATTR_OUTPUTS), metrics) .entrySet().stream().filter(ref -> !distinctOutputs.containsKey(ref.getKey())) .forEach(ref -> distinctOutputs.put(ref.getKey(), ref.getValue())); } final Referenceable flowPathRef = new Referenceable(flowPathGuid, TYPE_NIFI_FLOW_PATH, null); flowPathRef.set(ATTR_INPUTS, new ArrayList<>(distinctInputs.values())); flowPathRef.set(ATTR_OUTPUTS, new ArrayList<>(distinctOutputs.values())); return new HookNotification.EntityPartialUpdateRequest(NIFI_USER, TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName, flowPathRef); }) .filter(Objects::nonNull) .collect(Collectors.toList()); metrics.uniquePartialNiFiFlowPathUpdates = deduplicatedMessages.size(); notifier.accept(deduplicatedMessages); notifier.accept(otherMessages); } finally { logger.info(metrics.toLogString("Finished")); } }
NotificationSender { void send(final List<HookNotification.HookNotificationMessage> messages, final Consumer<List<HookNotification.HookNotificationMessage>> notifier) { final Metrics metrics = new Metrics(); try { metrics.totalMessages = messages.size(); final Map<Boolean, List<HookNotification.HookNotificationMessage>> createAndOthers = messages.stream().collect(groupingBy(msg -> ENTITY_CREATE.equals(msg.getType()))); final List<HookNotification.HookNotificationMessage> creates = safeGet(createAndOthers, true); metrics.createMessages = creates.size(); final Map<Boolean, List<Referenceable>> newFlowPathsAndOtherEntities = creates.stream() .flatMap(msg -> ((HookNotification.EntityCreateRequest) msg).getEntities().stream()) .collect(groupingBy(ref -> TYPE_NIFI_FLOW_PATH.equals(ref.typeName))); final List<Referenceable> newEntitiesExceptFlowPaths = safeGet(newFlowPathsAndOtherEntities, false) .stream().filter(distinctReferenceable()).collect(Collectors.toList()); final Collection<Referenceable> newFlowPaths = safeGet(newFlowPathsAndOtherEntities, true).stream() .collect(toMap(ref -> ref.get(ATTR_QUALIFIED_NAME), ref -> ref, (r1, r2) -> { mergeRefs(r1, r2); return r1; })).values(); metrics.uniqueFlowPathCreates = newFlowPaths.size(); metrics.uniqueOtherCreates = newEntitiesExceptFlowPaths.size(); List<Referenceable> newEntities = new ArrayList<>(); newEntities.addAll(newEntitiesExceptFlowPaths); newEntities.addAll(newFlowPaths); if (!newEntities.isEmpty()) { notifier.accept(Collections.singletonList(new HookNotification.EntityCreateRequest(NIFI_USER, newEntities))); } final Map<Boolean, List<HookNotification.HookNotificationMessage>> partialNiFiFlowPathUpdateAndOthers = safeGet(createAndOthers, false).stream().collect(groupingBy(msg -> ENTITY_PARTIAL_UPDATE.equals(msg.getType()) && TYPE_NIFI_FLOW_PATH.equals(((HookNotification.EntityPartialUpdateRequest)msg).getTypeName()) && ATTR_QUALIFIED_NAME.equals(((HookNotification.EntityPartialUpdateRequest)msg).getAttribute()) )); final List<HookNotification.HookNotificationMessage> partialNiFiFlowPathUpdates = safeGet(partialNiFiFlowPathUpdateAndOthers, true); final List<HookNotification.HookNotificationMessage> otherMessages = safeGet(partialNiFiFlowPathUpdateAndOthers, false); metrics.partialNiFiFlowPathUpdates = partialNiFiFlowPathUpdates.size(); metrics.otherMessages = otherMessages.size(); final List<HookNotification.HookNotificationMessage> deduplicatedMessages = partialNiFiFlowPathUpdates.stream().map(msg -> (HookNotification.EntityPartialUpdateRequest) msg) .collect(groupingBy(HookNotification.EntityPartialUpdateRequest::getAttributeValue)).entrySet().stream() .map(entry -> { final String flowPathQualifiedName = entry.getKey(); final Map<String, Referenceable> distinctInputs; final Map<String, Referenceable> distinctOutputs; final String flowPathGuid; try { metrics.flowPathSearched++; final AtlasEntity.AtlasEntityWithExtInfo flowPathExt = atlasClient.searchEntityDef(new AtlasObjectId(TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName)); final AtlasEntity flowPathEntity = flowPathExt.getEntity(); flowPathGuid = flowPathEntity.getGuid(); distinctInputs = toReferenceables(flowPathEntity.getAttribute(ATTR_INPUTS), metrics); distinctOutputs = toReferenceables(flowPathEntity.getAttribute(ATTR_OUTPUTS), metrics); } catch (AtlasServiceException e) { if (ClientResponse.Status.NOT_FOUND.equals(e.getStatus())) { logger.debug("nifi_flow_path was not found for qualifiedName {}", flowPathQualifiedName); } else { logger.warn("Failed to retrieve nifi_flow_path with qualifiedName {} due to {}", flowPathQualifiedName, e, e); } return null; } for (HookNotification.EntityPartialUpdateRequest msg : entry.getValue()) { fromReferenceable(msg.getEntity().get(ATTR_INPUTS), metrics) .entrySet().stream().filter(ref -> !distinctInputs.containsKey(ref.getKey())) .forEach(ref -> distinctInputs.put(ref.getKey(), ref.getValue())); fromReferenceable(msg.getEntity().get(ATTR_OUTPUTS), metrics) .entrySet().stream().filter(ref -> !distinctOutputs.containsKey(ref.getKey())) .forEach(ref -> distinctOutputs.put(ref.getKey(), ref.getValue())); } final Referenceable flowPathRef = new Referenceable(flowPathGuid, TYPE_NIFI_FLOW_PATH, null); flowPathRef.set(ATTR_INPUTS, new ArrayList<>(distinctInputs.values())); flowPathRef.set(ATTR_OUTPUTS, new ArrayList<>(distinctOutputs.values())); return new HookNotification.EntityPartialUpdateRequest(NIFI_USER, TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName, flowPathRef); }) .filter(Objects::nonNull) .collect(Collectors.toList()); metrics.uniquePartialNiFiFlowPathUpdates = deduplicatedMessages.size(); notifier.accept(deduplicatedMessages); notifier.accept(otherMessages); } finally { logger.info(metrics.toLogString("Finished")); } } }
NotificationSender { void send(final List<HookNotification.HookNotificationMessage> messages, final Consumer<List<HookNotification.HookNotificationMessage>> notifier) { final Metrics metrics = new Metrics(); try { metrics.totalMessages = messages.size(); final Map<Boolean, List<HookNotification.HookNotificationMessage>> createAndOthers = messages.stream().collect(groupingBy(msg -> ENTITY_CREATE.equals(msg.getType()))); final List<HookNotification.HookNotificationMessage> creates = safeGet(createAndOthers, true); metrics.createMessages = creates.size(); final Map<Boolean, List<Referenceable>> newFlowPathsAndOtherEntities = creates.stream() .flatMap(msg -> ((HookNotification.EntityCreateRequest) msg).getEntities().stream()) .collect(groupingBy(ref -> TYPE_NIFI_FLOW_PATH.equals(ref.typeName))); final List<Referenceable> newEntitiesExceptFlowPaths = safeGet(newFlowPathsAndOtherEntities, false) .stream().filter(distinctReferenceable()).collect(Collectors.toList()); final Collection<Referenceable> newFlowPaths = safeGet(newFlowPathsAndOtherEntities, true).stream() .collect(toMap(ref -> ref.get(ATTR_QUALIFIED_NAME), ref -> ref, (r1, r2) -> { mergeRefs(r1, r2); return r1; })).values(); metrics.uniqueFlowPathCreates = newFlowPaths.size(); metrics.uniqueOtherCreates = newEntitiesExceptFlowPaths.size(); List<Referenceable> newEntities = new ArrayList<>(); newEntities.addAll(newEntitiesExceptFlowPaths); newEntities.addAll(newFlowPaths); if (!newEntities.isEmpty()) { notifier.accept(Collections.singletonList(new HookNotification.EntityCreateRequest(NIFI_USER, newEntities))); } final Map<Boolean, List<HookNotification.HookNotificationMessage>> partialNiFiFlowPathUpdateAndOthers = safeGet(createAndOthers, false).stream().collect(groupingBy(msg -> ENTITY_PARTIAL_UPDATE.equals(msg.getType()) && TYPE_NIFI_FLOW_PATH.equals(((HookNotification.EntityPartialUpdateRequest)msg).getTypeName()) && ATTR_QUALIFIED_NAME.equals(((HookNotification.EntityPartialUpdateRequest)msg).getAttribute()) )); final List<HookNotification.HookNotificationMessage> partialNiFiFlowPathUpdates = safeGet(partialNiFiFlowPathUpdateAndOthers, true); final List<HookNotification.HookNotificationMessage> otherMessages = safeGet(partialNiFiFlowPathUpdateAndOthers, false); metrics.partialNiFiFlowPathUpdates = partialNiFiFlowPathUpdates.size(); metrics.otherMessages = otherMessages.size(); final List<HookNotification.HookNotificationMessage> deduplicatedMessages = partialNiFiFlowPathUpdates.stream().map(msg -> (HookNotification.EntityPartialUpdateRequest) msg) .collect(groupingBy(HookNotification.EntityPartialUpdateRequest::getAttributeValue)).entrySet().stream() .map(entry -> { final String flowPathQualifiedName = entry.getKey(); final Map<String, Referenceable> distinctInputs; final Map<String, Referenceable> distinctOutputs; final String flowPathGuid; try { metrics.flowPathSearched++; final AtlasEntity.AtlasEntityWithExtInfo flowPathExt = atlasClient.searchEntityDef(new AtlasObjectId(TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName)); final AtlasEntity flowPathEntity = flowPathExt.getEntity(); flowPathGuid = flowPathEntity.getGuid(); distinctInputs = toReferenceables(flowPathEntity.getAttribute(ATTR_INPUTS), metrics); distinctOutputs = toReferenceables(flowPathEntity.getAttribute(ATTR_OUTPUTS), metrics); } catch (AtlasServiceException e) { if (ClientResponse.Status.NOT_FOUND.equals(e.getStatus())) { logger.debug("nifi_flow_path was not found for qualifiedName {}", flowPathQualifiedName); } else { logger.warn("Failed to retrieve nifi_flow_path with qualifiedName {} due to {}", flowPathQualifiedName, e, e); } return null; } for (HookNotification.EntityPartialUpdateRequest msg : entry.getValue()) { fromReferenceable(msg.getEntity().get(ATTR_INPUTS), metrics) .entrySet().stream().filter(ref -> !distinctInputs.containsKey(ref.getKey())) .forEach(ref -> distinctInputs.put(ref.getKey(), ref.getValue())); fromReferenceable(msg.getEntity().get(ATTR_OUTPUTS), metrics) .entrySet().stream().filter(ref -> !distinctOutputs.containsKey(ref.getKey())) .forEach(ref -> distinctOutputs.put(ref.getKey(), ref.getValue())); } final Referenceable flowPathRef = new Referenceable(flowPathGuid, TYPE_NIFI_FLOW_PATH, null); flowPathRef.set(ATTR_INPUTS, new ArrayList<>(distinctInputs.values())); flowPathRef.set(ATTR_OUTPUTS, new ArrayList<>(distinctOutputs.values())); return new HookNotification.EntityPartialUpdateRequest(NIFI_USER, TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName, flowPathRef); }) .filter(Objects::nonNull) .collect(Collectors.toList()); metrics.uniquePartialNiFiFlowPathUpdates = deduplicatedMessages.size(); notifier.accept(deduplicatedMessages); notifier.accept(otherMessages); } finally { logger.info(metrics.toLogString("Finished")); } } NotificationSender(); }
NotificationSender { void send(final List<HookNotification.HookNotificationMessage> messages, final Consumer<List<HookNotification.HookNotificationMessage>> notifier) { final Metrics metrics = new Metrics(); try { metrics.totalMessages = messages.size(); final Map<Boolean, List<HookNotification.HookNotificationMessage>> createAndOthers = messages.stream().collect(groupingBy(msg -> ENTITY_CREATE.equals(msg.getType()))); final List<HookNotification.HookNotificationMessage> creates = safeGet(createAndOthers, true); metrics.createMessages = creates.size(); final Map<Boolean, List<Referenceable>> newFlowPathsAndOtherEntities = creates.stream() .flatMap(msg -> ((HookNotification.EntityCreateRequest) msg).getEntities().stream()) .collect(groupingBy(ref -> TYPE_NIFI_FLOW_PATH.equals(ref.typeName))); final List<Referenceable> newEntitiesExceptFlowPaths = safeGet(newFlowPathsAndOtherEntities, false) .stream().filter(distinctReferenceable()).collect(Collectors.toList()); final Collection<Referenceable> newFlowPaths = safeGet(newFlowPathsAndOtherEntities, true).stream() .collect(toMap(ref -> ref.get(ATTR_QUALIFIED_NAME), ref -> ref, (r1, r2) -> { mergeRefs(r1, r2); return r1; })).values(); metrics.uniqueFlowPathCreates = newFlowPaths.size(); metrics.uniqueOtherCreates = newEntitiesExceptFlowPaths.size(); List<Referenceable> newEntities = new ArrayList<>(); newEntities.addAll(newEntitiesExceptFlowPaths); newEntities.addAll(newFlowPaths); if (!newEntities.isEmpty()) { notifier.accept(Collections.singletonList(new HookNotification.EntityCreateRequest(NIFI_USER, newEntities))); } final Map<Boolean, List<HookNotification.HookNotificationMessage>> partialNiFiFlowPathUpdateAndOthers = safeGet(createAndOthers, false).stream().collect(groupingBy(msg -> ENTITY_PARTIAL_UPDATE.equals(msg.getType()) && TYPE_NIFI_FLOW_PATH.equals(((HookNotification.EntityPartialUpdateRequest)msg).getTypeName()) && ATTR_QUALIFIED_NAME.equals(((HookNotification.EntityPartialUpdateRequest)msg).getAttribute()) )); final List<HookNotification.HookNotificationMessage> partialNiFiFlowPathUpdates = safeGet(partialNiFiFlowPathUpdateAndOthers, true); final List<HookNotification.HookNotificationMessage> otherMessages = safeGet(partialNiFiFlowPathUpdateAndOthers, false); metrics.partialNiFiFlowPathUpdates = partialNiFiFlowPathUpdates.size(); metrics.otherMessages = otherMessages.size(); final List<HookNotification.HookNotificationMessage> deduplicatedMessages = partialNiFiFlowPathUpdates.stream().map(msg -> (HookNotification.EntityPartialUpdateRequest) msg) .collect(groupingBy(HookNotification.EntityPartialUpdateRequest::getAttributeValue)).entrySet().stream() .map(entry -> { final String flowPathQualifiedName = entry.getKey(); final Map<String, Referenceable> distinctInputs; final Map<String, Referenceable> distinctOutputs; final String flowPathGuid; try { metrics.flowPathSearched++; final AtlasEntity.AtlasEntityWithExtInfo flowPathExt = atlasClient.searchEntityDef(new AtlasObjectId(TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName)); final AtlasEntity flowPathEntity = flowPathExt.getEntity(); flowPathGuid = flowPathEntity.getGuid(); distinctInputs = toReferenceables(flowPathEntity.getAttribute(ATTR_INPUTS), metrics); distinctOutputs = toReferenceables(flowPathEntity.getAttribute(ATTR_OUTPUTS), metrics); } catch (AtlasServiceException e) { if (ClientResponse.Status.NOT_FOUND.equals(e.getStatus())) { logger.debug("nifi_flow_path was not found for qualifiedName {}", flowPathQualifiedName); } else { logger.warn("Failed to retrieve nifi_flow_path with qualifiedName {} due to {}", flowPathQualifiedName, e, e); } return null; } for (HookNotification.EntityPartialUpdateRequest msg : entry.getValue()) { fromReferenceable(msg.getEntity().get(ATTR_INPUTS), metrics) .entrySet().stream().filter(ref -> !distinctInputs.containsKey(ref.getKey())) .forEach(ref -> distinctInputs.put(ref.getKey(), ref.getValue())); fromReferenceable(msg.getEntity().get(ATTR_OUTPUTS), metrics) .entrySet().stream().filter(ref -> !distinctOutputs.containsKey(ref.getKey())) .forEach(ref -> distinctOutputs.put(ref.getKey(), ref.getValue())); } final Referenceable flowPathRef = new Referenceable(flowPathGuid, TYPE_NIFI_FLOW_PATH, null); flowPathRef.set(ATTR_INPUTS, new ArrayList<>(distinctInputs.values())); flowPathRef.set(ATTR_OUTPUTS, new ArrayList<>(distinctOutputs.values())); return new HookNotification.EntityPartialUpdateRequest(NIFI_USER, TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName, flowPathRef); }) .filter(Objects::nonNull) .collect(Collectors.toList()); metrics.uniquePartialNiFiFlowPathUpdates = deduplicatedMessages.size(); notifier.accept(deduplicatedMessages); notifier.accept(otherMessages); } finally { logger.info(metrics.toLogString("Finished")); } } NotificationSender();  }
NotificationSender { void send(final List<HookNotification.HookNotificationMessage> messages, final Consumer<List<HookNotification.HookNotificationMessage>> notifier) { final Metrics metrics = new Metrics(); try { metrics.totalMessages = messages.size(); final Map<Boolean, List<HookNotification.HookNotificationMessage>> createAndOthers = messages.stream().collect(groupingBy(msg -> ENTITY_CREATE.equals(msg.getType()))); final List<HookNotification.HookNotificationMessage> creates = safeGet(createAndOthers, true); metrics.createMessages = creates.size(); final Map<Boolean, List<Referenceable>> newFlowPathsAndOtherEntities = creates.stream() .flatMap(msg -> ((HookNotification.EntityCreateRequest) msg).getEntities().stream()) .collect(groupingBy(ref -> TYPE_NIFI_FLOW_PATH.equals(ref.typeName))); final List<Referenceable> newEntitiesExceptFlowPaths = safeGet(newFlowPathsAndOtherEntities, false) .stream().filter(distinctReferenceable()).collect(Collectors.toList()); final Collection<Referenceable> newFlowPaths = safeGet(newFlowPathsAndOtherEntities, true).stream() .collect(toMap(ref -> ref.get(ATTR_QUALIFIED_NAME), ref -> ref, (r1, r2) -> { mergeRefs(r1, r2); return r1; })).values(); metrics.uniqueFlowPathCreates = newFlowPaths.size(); metrics.uniqueOtherCreates = newEntitiesExceptFlowPaths.size(); List<Referenceable> newEntities = new ArrayList<>(); newEntities.addAll(newEntitiesExceptFlowPaths); newEntities.addAll(newFlowPaths); if (!newEntities.isEmpty()) { notifier.accept(Collections.singletonList(new HookNotification.EntityCreateRequest(NIFI_USER, newEntities))); } final Map<Boolean, List<HookNotification.HookNotificationMessage>> partialNiFiFlowPathUpdateAndOthers = safeGet(createAndOthers, false).stream().collect(groupingBy(msg -> ENTITY_PARTIAL_UPDATE.equals(msg.getType()) && TYPE_NIFI_FLOW_PATH.equals(((HookNotification.EntityPartialUpdateRequest)msg).getTypeName()) && ATTR_QUALIFIED_NAME.equals(((HookNotification.EntityPartialUpdateRequest)msg).getAttribute()) )); final List<HookNotification.HookNotificationMessage> partialNiFiFlowPathUpdates = safeGet(partialNiFiFlowPathUpdateAndOthers, true); final List<HookNotification.HookNotificationMessage> otherMessages = safeGet(partialNiFiFlowPathUpdateAndOthers, false); metrics.partialNiFiFlowPathUpdates = partialNiFiFlowPathUpdates.size(); metrics.otherMessages = otherMessages.size(); final List<HookNotification.HookNotificationMessage> deduplicatedMessages = partialNiFiFlowPathUpdates.stream().map(msg -> (HookNotification.EntityPartialUpdateRequest) msg) .collect(groupingBy(HookNotification.EntityPartialUpdateRequest::getAttributeValue)).entrySet().stream() .map(entry -> { final String flowPathQualifiedName = entry.getKey(); final Map<String, Referenceable> distinctInputs; final Map<String, Referenceable> distinctOutputs; final String flowPathGuid; try { metrics.flowPathSearched++; final AtlasEntity.AtlasEntityWithExtInfo flowPathExt = atlasClient.searchEntityDef(new AtlasObjectId(TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName)); final AtlasEntity flowPathEntity = flowPathExt.getEntity(); flowPathGuid = flowPathEntity.getGuid(); distinctInputs = toReferenceables(flowPathEntity.getAttribute(ATTR_INPUTS), metrics); distinctOutputs = toReferenceables(flowPathEntity.getAttribute(ATTR_OUTPUTS), metrics); } catch (AtlasServiceException e) { if (ClientResponse.Status.NOT_FOUND.equals(e.getStatus())) { logger.debug("nifi_flow_path was not found for qualifiedName {}", flowPathQualifiedName); } else { logger.warn("Failed to retrieve nifi_flow_path with qualifiedName {} due to {}", flowPathQualifiedName, e, e); } return null; } for (HookNotification.EntityPartialUpdateRequest msg : entry.getValue()) { fromReferenceable(msg.getEntity().get(ATTR_INPUTS), metrics) .entrySet().stream().filter(ref -> !distinctInputs.containsKey(ref.getKey())) .forEach(ref -> distinctInputs.put(ref.getKey(), ref.getValue())); fromReferenceable(msg.getEntity().get(ATTR_OUTPUTS), metrics) .entrySet().stream().filter(ref -> !distinctOutputs.containsKey(ref.getKey())) .forEach(ref -> distinctOutputs.put(ref.getKey(), ref.getValue())); } final Referenceable flowPathRef = new Referenceable(flowPathGuid, TYPE_NIFI_FLOW_PATH, null); flowPathRef.set(ATTR_INPUTS, new ArrayList<>(distinctInputs.values())); flowPathRef.set(ATTR_OUTPUTS, new ArrayList<>(distinctOutputs.values())); return new HookNotification.EntityPartialUpdateRequest(NIFI_USER, TYPE_NIFI_FLOW_PATH, ATTR_QUALIFIED_NAME, flowPathQualifiedName, flowPathRef); }) .filter(Objects::nonNull) .collect(Collectors.toList()); metrics.uniquePartialNiFiFlowPathUpdates = deduplicatedMessages.size(); notifier.accept(deduplicatedMessages); notifier.accept(otherMessages); } finally { logger.info(metrics.toLogString("Finished")); } } NotificationSender();   }