ClusterAlertMessaging implements AlertMessaging { @Override public void sendAlert(final DistributedMember member, final AlertLevel alertLevel, final Instant timestamp, final String threadName, final long threadId, final String formattedMessage, final String stackTrace) { executor.submit(() -> AlertingAction.execute(() -> { try { String connectionName = system.getConfig().getName(); AlertListenerMessage message = alertListenerMessageFactory.createAlertListenerMessage(member, alertLevel, timestamp, connectionName, threadName, threadId, formattedMessage, stackTrace); if (member.equals(system.getDistributedMember())) { LOGGER.debug("Processing local alert message: {}, {}, {}, {}, {}, {}, [{}], [{}].", member, alertLevel, timestamp, connectionName, threadName, threadId, formattedMessage, stackTrace); processAlertListenerMessage(message); } else { LOGGER.debug("Sending remote alert message: {}, {}, {}, {}, {}, {}, [{}], [{}].", member, alertLevel, timestamp, connectionName, threadName, threadId, formattedMessage, stackTrace); dm.putOutgoing(message); } } catch (ReenteredConnectException ignore) { } catch (AlertingIOException e) { alertingIOExceptionLogger.accept(e); } })); } ClusterAlertMessaging(final InternalDistributedSystem system); @VisibleForTesting  ClusterAlertMessaging(final InternalDistributedSystem system, final DistributionManager dm,
      final AlertListenerMessageFactory alertListenerMessageFactory,
      final ExecutorService executor); @VisibleForTesting  ClusterAlertMessaging(final InternalDistributedSystem system, final DistributionManager dm,
      final AlertListenerMessageFactory alertListenerMessageFactory, final ExecutorService executor,
      final Consumer<AlertingIOException> alertingIOExceptionLogger); @Override void sendAlert(final DistributedMember member,
      final AlertLevel alertLevel,
      final Instant timestamp,
      final String threadName,
      final long threadId,
      final String formattedMessage,
      final String stackTrace); void close(); @Override String toString();  }
@Test public void sendAlertSendsMessageIfMemberIsRemote() { DistributionManager distributionManager = mock(ClusterDistributionManager.class); ClusterAlertMessaging clusterAlertMessaging = spyClusterAlertMessaging(distributionManager, currentThreadExecutorService()); clusterAlertMessaging.sendAlert(remoteMember, AlertLevel.WARNING, Instant.now(), "threadName", Thread.currentThread().getId(), "formattedMessage", "stackTrace"); verify(distributionManager).putOutgoing(eq(alertListenerMessage)); }
@Test public void sendAlertUsesExecutorService() { ExecutorService executor = currentThreadExecutorService(); ClusterAlertMessaging clusterAlertMessaging = spyClusterAlertMessaging(mock(ClusterDistributionManager.class), executor); clusterAlertMessaging.sendAlert(remoteMember, AlertLevel.WARNING, Instant.now(), "threadName", Thread.currentThread().getId(), "formattedMessage", "stackTrace"); verify(executor).submit(any(Runnable.class)); }
@Test public void sendAlertUsesAlertingAction() { ExecutorService executor = currentThreadExecutorService(); ClusterDistributionManager distributionManager = mock(ClusterDistributionManager.class); ClusterAlertMessaging clusterAlertMessaging = spyClusterAlertMessaging(distributionManager, executor); when(distributionManager.putOutgoing(any())) .thenAnswer(invocation -> { assertThat(AlertingAction.isThreadAlerting()).isTrue(); return null; }); clusterAlertMessaging.sendAlert(remoteMember, AlertLevel.WARNING, Instant.now(), "threadName", Thread.currentThread().getId(), "formattedMessage", "stackTrace"); verify(distributionManager).putOutgoing(any()); }
@Test public void sendAlertLogsWarning_ifAlertingIOExceptionIsCaught() { ExecutorService executor = currentThreadExecutorService(); ClusterDistributionManager distributionManager = mock(ClusterDistributionManager.class); Consumer<AlertingIOException> alertingIOExceptionLogger = uncheckedCast(mock(Consumer.class)); ClusterAlertMessaging clusterAlertMessaging = spyClusterAlertMessaging(distributionManager, executor, alertingIOExceptionLogger); doThrow(new AlertingIOException(new IOException("Cannot form connection to alert listener"))) .when(distributionManager).putOutgoing(any()); clusterAlertMessaging.sendAlert(remoteMember, AlertLevel.WARNING, Instant.now(), "threadName", Thread.currentThread().getId(), "formattedMessage", "stackTrace"); ArgumentCaptor<AlertingIOException> captor = ArgumentCaptor.forClass(AlertingIOException.class); verify(alertingIOExceptionLogger).accept(captor.capture()); assertThat(captor.getValue()) .isInstanceOf(AlertingIOException.class) .hasMessageContaining("Cannot form connection to alert listener"); }
@Test public void sendAlertLogsWarningOnce_ifAlertingIOExceptionIsCaught() { ExecutorService executor = currentThreadExecutorService(); ClusterDistributionManager distributionManager = mock(ClusterDistributionManager.class); Consumer<AlertingIOException> alertingIOExceptionLogger = uncheckedCast(mock(Consumer.class)); ClusterAlertMessaging clusterAlertMessaging = spyClusterAlertMessaging(distributionManager, executor, alertingIOExceptionLogger); doThrow(new AlertingIOException(new IOException("Cannot form connection to alert listener"))) .when(distributionManager).putOutgoing(any()); clusterAlertMessaging.sendAlert(remoteMember, AlertLevel.WARNING, Instant.now(), "threadName", Thread.currentThread().getId(), "formattedMessage", "stackTrace"); ArgumentCaptor<AlertingIOException> captor = ArgumentCaptor.forClass(AlertingIOException.class); verify(alertingIOExceptionLogger).accept(captor.capture()); assertThat(captor.getAllValues()).hasSize(1); }
@Test public void sendAlertDoesNotSend_ifAlertingIOExceptionIsCaught() { ExecutorService executor = currentThreadExecutorService(); ClusterDistributionManager distributionManager = mock(ClusterDistributionManager.class); ClusterAlertMessaging clusterAlertMessaging = spyClusterAlertMessaging(distributionManager, executor); when(distributionManager.putOutgoing(any())) .thenAnswer(invocation -> { assertThat(AlertingAction.isThreadAlerting()).isTrue(); return null; }); clusterAlertMessaging.sendAlert(remoteMember, AlertLevel.WARNING, Instant.now(), "threadName", Thread.currentThread().getId(), "formattedMessage", "stackTrace"); verifyNoMoreInteractions(distributionManager); }