AtlasHook { @VisibleForTesting static void notifyEntitiesInternal(List<HookNotification.HookNotificationMessage> messages, int maxRetries, NotificationInterface notificationInterface, boolean shouldLogFailedMessages, FailedMessagesLogger logger) { if (messages == null || messages.isEmpty()) { return; } final String message = messages.toString(); int numRetries = 0; while (true) { try { notificationInterface.send(NotificationInterface.NotificationType.HOOK, messages); return; } catch (Exception e) { numRetries++; if (numRetries < maxRetries) { LOG.error("Failed to send notification - attempt #{}; error={}", numRetries, e.getMessage()); try { LOG.debug("Sleeping for {} ms before retry", notificationRetryInterval); Thread.sleep(notificationRetryInterval); } catch (InterruptedException ie) { LOG.error("Notification hook thread sleep interrupted"); } } else { if (shouldLogFailedMessages && e instanceof NotificationException) { List<String> failedMessages = ((NotificationException) e).getFailedMessages(); for (String msg : failedMessages) { logger.log(msg); } } LOG.error("Failed to notify atlas for entity {} after {} retries. Quitting", message, maxRetries, e); return; } } } }  static void notifyEntities(List<HookNotification.HookNotificationMessage> messages, int maxRetries); static String getUser(); static String getUser(String userName); static String getUser(String userName, UserGroupInformation ugi); static final String ATLAS_NOTIFICATION_RETRY_INTERVAL; static final String ATLAS_NOTIFICATION_FAILED_MESSAGES_FILENAME_KEY; static final String ATLAS_HOOK_FAILED_MESSAGES_LOG_DEFAULT_NAME; static final String ATLAS_NOTIFICATION_LOG_FAILED_MESSAGES_ENABLED_KEY; }
@Test (timeOut = 10000) public void testNotifyEntitiesDoesNotHangOnException() throws Exception { List<HookNotification.HookNotificationMessage> hookNotificationMessages = new ArrayList<>(); doThrow(new NotificationException(new Exception())).when(notificationInterface) .send(NotificationInterface.NotificationType.HOOK, hookNotificationMessages); AtlasHook.notifyEntitiesInternal(hookNotificationMessages, 0, notificationInterface, false, failedMessagesLogger); }
@Test public void testNotifyEntitiesRetriesOnException() throws NotificationException { List<HookNotification.HookNotificationMessage> hookNotificationMessages = new ArrayList<HookNotification.HookNotificationMessage>() {{ add(new HookNotification.EntityCreateRequest("user")); } }; doThrow(new NotificationException(new Exception())).when(notificationInterface) .send(NotificationInterface.NotificationType.HOOK, hookNotificationMessages); AtlasHook.notifyEntitiesInternal(hookNotificationMessages, 2, notificationInterface, false, failedMessagesLogger); verify(notificationInterface, times(2)). send(NotificationInterface.NotificationType.HOOK, hookNotificationMessages); }
@Test public void testFailedMessageIsLoggedIfRequired() throws NotificationException { List<HookNotification.HookNotificationMessage> hookNotificationMessages = new ArrayList<HookNotification.HookNotificationMessage>() {{ add(new HookNotification.EntityCreateRequest("user")); } }; doThrow(new NotificationException(new Exception(), Arrays.asList("test message"))) .when(notificationInterface) .send(NotificationInterface.NotificationType.HOOK, hookNotificationMessages); AtlasHook.notifyEntitiesInternal(hookNotificationMessages, 2, notificationInterface, true, failedMessagesLogger); verify(failedMessagesLogger, times(1)).log("test message"); }
@Test public void testFailedMessageIsNotLoggedIfNotRequired() throws NotificationException { List<HookNotification.HookNotificationMessage> hookNotificationMessages = new ArrayList<>(); doThrow(new NotificationException(new Exception(), Arrays.asList("test message"))) .when(notificationInterface) .send(NotificationInterface.NotificationType.HOOK, hookNotificationMessages); AtlasHook.notifyEntitiesInternal(hookNotificationMessages, 2, notificationInterface, false, failedMessagesLogger); verifyZeroInteractions(failedMessagesLogger); }
@Test public void testAllFailedMessagesAreLogged() throws NotificationException { List<HookNotification.HookNotificationMessage> hookNotificationMessages = new ArrayList<HookNotification.HookNotificationMessage>() {{ add(new HookNotification.EntityCreateRequest("user")); } }; doThrow(new NotificationException(new Exception(), Arrays.asList("test message1", "test message2"))) .when(notificationInterface) .send(NotificationInterface.NotificationType.HOOK, hookNotificationMessages); AtlasHook.notifyEntitiesInternal(hookNotificationMessages, 2, notificationInterface, true, failedMessagesLogger); verify(failedMessagesLogger, times(1)).log("test message1"); verify(failedMessagesLogger, times(1)).log("test message2"); }
@Test public void testFailedMessageIsNotLoggedIfNotANotificationException() throws Exception { List<HookNotification.HookNotificationMessage> hookNotificationMessages = new ArrayList<>(); doThrow(new RuntimeException("test message")).when(notificationInterface) .send(NotificationInterface.NotificationType.HOOK, hookNotificationMessages); AtlasHook.notifyEntitiesInternal(hookNotificationMessages, 2, notificationInterface, true, failedMessagesLogger); verifyZeroInteractions(failedMessagesLogger); }