@VisibleForTesting()
    void checkDiskUsage() {
        final Map<Notification.Type, List<String>> notificationTypePerNodeIdentifier = new HashMap<>();
        try {
            ClusterAllocationDiskSettings settings = cluster.getClusterAllocationDiskSettings();
            if (settings.ThresholdEnabled()) {
                final Set<NodeDiskUsageStats> diskUsageStats = cluster.getDiskUsageStats();

                for (NodeDiskUsageStats nodeDiskUsageStats : diskUsageStats) {
                    if (!nodeHoldsData(nodeDiskUsageStats)) {
                        LOG.debug("Ignoring non-data ES node <{}/{}> with roles <{}> for disk usage check.",
                                nodeDiskUsageStats.host(), nodeDiskUsageStats.ip(), nodeDiskUsageStats.roles());
                        continue;
                    }
                    Notification.Type currentNodeNotificationType = null;
                    WatermarkSettings<?> watermarkSettings = settings.watermarkSettings();
                    if (watermarkSettings instanceof PercentageWatermarkSettings) {
                        currentNodeNotificationType = getDiskUsageNotificationTypeByPercentage((PercentageWatermarkSettings) watermarkSettings, nodeDiskUsageStats);
                    } else if (watermarkSettings instanceof AbsoluteValueWatermarkSettings) {
                        currentNodeNotificationType = getDiskUsageNotificationTypeByAbsoluteValues((AbsoluteValueWatermarkSettings) watermarkSettings, nodeDiskUsageStats);
                    }
                    if (currentNodeNotificationType != null) {
                        String nodeIdentifier = firstNonNull(nodeDiskUsageStats.host(), nodeDiskUsageStats.ip());
                        notificationTypePerNodeIdentifier.merge(currentNodeNotificationType, Collections.singletonList(nodeIdentifier), (prev, cur) -> ImmutableList.<String>builder()
                                .addAll(prev)
                                .addAll(cur)
                                .build());
                    }
                }

                if (notificationTypePerNodeIdentifier.isEmpty()) {
                    fixAllDiskUsageNotifications();
                } else {
                    publishDiskUsageNotifications(notificationTypePerNodeIdentifier);
                }
            }
        } catch (Exception e) {
            LOG.error("Error while trying to check Elasticsearch disk usage.Details: " + e.getMessage());
        }
    }
@Test
    public void fixAllDiskUsageNotificationsPercentage() throws Exception {
        Set<NodeDiskUsageStats> nodeDiskUsageStats = mockNodeDiskUsageStats();
        when(cluster.getDiskUsageStats()).thenReturn(nodeDiskUsageStats);
        when(cluster.getClusterAllocationDiskSettings()).thenReturn(buildThresholdNotTriggeredClusterAllocationDiskSettings(WatermarkSettings.SettingsType.PERCENTAGE));

        indexerClusterCheckerThread.checkDiskUsage();

        verify(notificationService, never()).publishIfFirst(any());
        verify(notificationService, times(1)).fixed(Notification.Type.ES_NODE_DISK_WATERMARK_FLOOD_STAGE);
        verify(notificationService, times(1)).fixed(Notification.Type.ES_NODE_DISK_WATERMARK_HIGH);
        verify(notificationService, times(1)).fixed(ES_NODE_DISK_WATERMARK_LOW);
    }