class 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)));

            // Deduplicate same entity creation messages.
            final List<Referenceable> newEntitiesExceptFlowPaths = safeGet(newFlowPathsAndOtherEntities, false)
                    .stream().filter(distinctReferenceable()).collect(Collectors.toList());

            // Deduplicate same flow paths and also merge inputs and outputs
            final Collection<Referenceable> newFlowPaths = safeGet(newFlowPathsAndOtherEntities, true).stream()
                    .collect(toMap(ref -> ref.get(ATTR_QUALIFIED_NAME), ref -> ref, (r1, r2) -> {
                        // Merge inputs and outputs.
                        mergeRefs(r1, r2);
                        return r1;
                    })).values();
            metrics.uniqueFlowPathCreates = newFlowPaths.size();
            metrics.uniqueOtherCreates = newEntitiesExceptFlowPaths.size();


            // 1-1. Notify new entities except 'nifi_flow_path'
            // 1-2. Notify new 'nifi_flow_path'
            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())
            ));


            // These updates are made against existing flow path entities.
            final List<HookNotification.HookNotificationMessage> partialNiFiFlowPathUpdates = safeGet(partialNiFiFlowPathUpdateAndOthers, true);
            final List<HookNotification.HookNotificationMessage> otherMessages = safeGet(partialNiFiFlowPathUpdateAndOthers, false);
            metrics.partialNiFiFlowPathUpdates = partialNiFiFlowPathUpdates.size();
            metrics.otherMessages = otherMessages.size();


            // 2. Notify de-duplicated 'nifi_flow_path' updates
            final List<HookNotification.HookNotificationMessage> deduplicatedMessages = partialNiFiFlowPathUpdates.stream().map(msg -> (HookNotification.EntityPartialUpdateRequest) msg)
                    // Group by nifi_flow_path qualifiedName value.
                    .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 {
                            // Fetch existing nifi_flow_path and its inputs/ouputs.
                            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;
                        }

                        // Merge all inputs and outputs for this nifi_flow_path.
                        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()));
                        }

                        // Consolidate messages into one.
                        final Referenceable flowPathRef = new Referenceable(flowPathGuid, TYPE_NIFI_FLOW_PATH, null);
                        // NOTE: distinctInputs.values() returns HashMap$Values, which causes following error. To avoid that, wrap with ArrayList:
                        // org.json4s.package$MappingException: Can't find ScalaSig for class org.apache.atlas.typesystem.Referenceable
                        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);

            // 3. Notify other messages
            notifier.accept(otherMessages);

        } finally {
            logger.info(metrics.toLogString("Finished"));
        }

    }

      NotificationSender();

}