geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kl...@apache.org
Subject [geode] branch develop updated: GEODE-5923: Add test for local alerts and cleanup classes (#2710)
Date Thu, 25 Oct 2018 17:58:25 GMT
This is an automated email from the ASF dual-hosted git repository.

klund pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git


The following commit(s) were added to refs/heads/develop by this push:
     new a51c036  GEODE-5923: Add test for local alerts and cleanup classes (#2710)
a51c036 is described below

commit a51c03684383335c6ebce00c9d6af17a8170413e
Author: Kirk Lund <klund@apache.org>
AuthorDate: Thu Oct 25 10:58:17 2018 -0700

    GEODE-5923: Add test for local alerts and cleanup classes (#2710)
    
    * Create DistributedSystemMXBeanIntegrationTest
    * Add ignored test that confirms bug GEODE-5923
    * Cleanup alert related classes and add lots of comments about how they work
---
 .../DistributedSystemMXBeanIntegrationTest.java    | 134 +++++++++++
 .../admin/remote/AlertLevelChangeMessage.java      |  30 +--
 .../admin/remote/AlertListenerMessage.java         | 107 +++++----
 .../management/internal/FederatingManager.java     | 257 +++++++++++----------
 .../management/internal/ManagerStartupMessage.java |  33 +--
 .../geode/management/internal/MemberMessenger.java |  78 +++++--
 6 files changed, 407 insertions(+), 232 deletions(-)

diff --git a/geode-core/src/integrationTest/java/org/apache/geode/management/DistributedSystemMXBeanIntegrationTest.java
b/geode-core/src/integrationTest/java/org/apache/geode/management/DistributedSystemMXBeanIntegrationTest.java
new file mode 100644
index 0000000..fefc923
--- /dev/null
+++ b/geode-core/src/integrationTest/java/org/apache/geode/management/DistributedSystemMXBeanIntegrationTest.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.geode.management;
+
+import static java.lang.management.ManagementFactory.getPlatformMBeanServer;
+import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_TIME_STATISTICS;
+import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_PORT;
+import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER;
+import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_PORT;
+import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_START;
+import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
+import static org.apache.geode.distributed.ConfigurationProperties.NAME;
+import static org.apache.geode.distributed.ConfigurationProperties.STATISTIC_SAMPLING_ENABLED;
+import static org.apache.geode.management.internal.MBeanJMXAdapter.getDistributedSystemName;
+import static org.apache.geode.test.dunit.standalone.DUnitLauncher.getDistributedSystemProperties;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+import java.util.Properties;
+
+import javax.management.JMX;
+import javax.management.Notification;
+import javax.management.NotificationFilter;
+import javax.management.NotificationListener;
+
+import org.apache.logging.log4j.Logger;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.distributed.DistributedSystem;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.test.junit.categories.ManagementTest;
+
+/**
+ * Integration tests for {@link DistributedSystemMXBean} with local {@link DistributedSystem}
+ * connection.
+ */
+@Category(ManagementTest.class)
+public class DistributedSystemMXBeanIntegrationTest {
+
+  private String name;
+  private InternalCache cache;
+  private Logger logger;
+  private String alertMessage;
+
+  private DistributedSystemMXBean distributedSystemMXBean;
+
+  @Rule
+  public TestName testName = new TestName();
+
+  @Before
+  public void setUp() throws Exception {
+    name = "Manager in " + testName.getMethodName();
+
+    Properties config = getDistributedSystemProperties();
+    config.setProperty(NAME, name);
+
+    config.setProperty(LOCATORS, "");
+    config.setProperty(JMX_MANAGER, "true");
+    config.setProperty(JMX_MANAGER_START, "true");
+    config.setProperty(JMX_MANAGER_PORT, "0");
+    config.setProperty(HTTP_SERVICE_PORT, "0");
+    config.setProperty(ENABLE_TIME_STATISTICS, "true");
+    config.setProperty(STATISTIC_SAMPLING_ENABLED, "true");
+
+    cache = (InternalCache) new CacheFactory(config).create();
+    logger = LogService.getLogger();
+
+    alertMessage = "Alerting in " + testName.getMethodName();
+
+    distributedSystemMXBean = JMX.newMXBeanProxy(getPlatformMBeanServer(),
+        getDistributedSystemName(), DistributedSystemMXBean.class);
+  }
+
+  @After
+  public void tearDown() {
+    cache.close();
+  }
+
+  @Test
+  public void isRegisteredInManager() {
+    assertThat(getPlatformMBeanServer().isRegistered(getDistributedSystemName())).isTrue();
+  }
+
+  @Test
+  public void supportsJmxProxy() {
+    assertThat(distributedSystemMXBean).isNotNull();
+    assertThat(distributedSystemMXBean.listMembers()).containsExactly(name);
+  }
+
+  /**
+   * This test confirms existence of bug GEODE-5923.
+   */
+  @Test
+  @Ignore("GEODE-5923")
+  public void providesSystemAlertNotification() throws Exception {
+    NotificationListener notificationListener = spy(NotificationListener.class);
+    NotificationFilter notificationFilter = (Notification notification) -> notification.getType()
+        .equals(JMXNotificationType.SYSTEM_ALERT);
+    getPlatformMBeanServer().addNotificationListener(getDistributedSystemName(),
+        notificationListener, notificationFilter, null);
+
+    // work around GEODE-5924 by invoking DistributedSystemMXBean.changeAlertLevel
+    distributedSystemMXBean.changeAlertLevel("warning");
+
+    logger.fatal(alertMessage);
+
+    verify(notificationListener).handleNotification(isA(Notification.class), isNull());
+  }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertLevelChangeMessage.java
b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertLevelChangeMessage.java
index 8ada228..1a950e6 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertLevelChangeMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertLevelChangeMessage.java
@@ -32,8 +32,6 @@ import org.apache.geode.internal.logging.log4j.LogMarker;
  * A message that is sent to make members of the distributed system aware that a manager
agent wants
  * alerts at a new level.
  *
- * @see AlertLevel
- *
  * @since GemFire 3.5
  */
 public class AlertLevelChangeMessage extends SerialDistributionMessage {
@@ -43,31 +41,28 @@ public class AlertLevelChangeMessage extends SerialDistributionMessage
{
   /** The new alert level */
   private int newLevel;
 
-  /////////////////////// Static Methods ///////////////////////
-
   /**
-   * Creates a new <code>AlertLevelChangeMessage</code>
+   * Creates a new {@code AlertLevelChangeMessage}.
    */
   public static AlertLevelChangeMessage create(int newLevel) {
-    AlertLevelChangeMessage m = new AlertLevelChangeMessage();
-    m.newLevel = newLevel;
-    return m;
+    AlertLevelChangeMessage alertLevelChangeMessage = new AlertLevelChangeMessage();
+    alertLevelChangeMessage.newLevel = newLevel;
+    return alertLevelChangeMessage;
   }
 
-  ////////////////////// Instance Methods //////////////////////
-
   @Override
   public void process(ClusterDistributionManager dm) {
-    AlertAppender.getInstance().removeAlertListener(this.getSender());
+    AlertAppender.getInstance().removeAlertListener(getSender());
 
-    if (this.newLevel != Alert.OFF) {
-      AlertAppender.getInstance().addAlertListener(this.getSender(), this.newLevel);
+    if (newLevel != Alert.OFF) {
+      AlertAppender.getInstance().addAlertListener(getSender(), newLevel);
       if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
-        logger.trace(LogMarker.DM_VERBOSE, "Added new AlertListener to application log writer");
+        logger.trace(LogMarker.DM_VERBOSE, "Added new AlertListener");
       }
     }
   }
 
+  @Override
   public int getDSFID() {
     return ALERT_LEVEL_CHANGE_MESSAGE;
   }
@@ -75,18 +70,17 @@ public class AlertLevelChangeMessage extends SerialDistributionMessage
{
   @Override
   public void toData(DataOutput out) throws IOException {
     super.toData(out);
-    out.writeInt(this.newLevel);
+    out.writeInt(newLevel);
   }
 
   @Override
   public void fromData(DataInput in) throws IOException, ClassNotFoundException {
     super.fromData(in);
-    this.newLevel = in.readInt();
+    newLevel = in.readInt();
   }
 
   @Override
   public String toString() {
-    return String.format("Changing alert level to %s",
-        AlertLevel.forSeverity(this.newLevel));
+    return String.format("Changing alert level to %s", AlertLevel.forSeverity(newLevel));
   }
 }
diff --git a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertListenerMessage.java
b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertListenerMessage.java
index 0080700..6c3b16b 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertListenerMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/AlertListenerMessage.java
@@ -12,8 +12,6 @@
  * or implied. See the License for the specific language governing permissions and limitations
under
  * the License.
  */
-
-
 package org.apache.geode.internal.admin.remote;
 
 import java.io.DataInput;
@@ -23,6 +21,7 @@ import java.util.Date;
 
 import org.apache.geode.DataSerializer;
 import org.apache.geode.admin.AlertLevel;
+import org.apache.geode.distributed.DistributedMember;
 import org.apache.geode.distributed.internal.AdminMessageType;
 import org.apache.geode.distributed.internal.ClusterDistributionManager;
 import org.apache.geode.distributed.internal.PooledDistributionMessage;
@@ -32,63 +31,71 @@ import org.apache.geode.internal.admin.Alert;
 import org.apache.geode.management.internal.AlertDetails;
 
 /**
- * A message that is sent to a particular console distribution manager to notify it of an
alert.
+ * A message that is sent to an admin member or manager to notify it of an alert.
+ *
+ * <p>
+ * Alerts are sent from remote members to the manager via {@link AlertListenerMessage} which
is
+ * no-ack (asynchronous). This means you cannot log a warning in one VM and immediately verify
that
+ * it arrives in the manager VM. You have to use {@code Mockito#timeout(long)} or
+ * {@code Awaitility}.
  */
 public class AlertListenerMessage extends PooledDistributionMessage implements AdminMessageType
{
-  // instance variables
-  private int msgLevel;
-  private Date msgDate;
+
+  private int alertLevel;
+  private Date date;
   private String connectionName;
   private String threadName;
-  private long tid;
-  private String msg;
+  private long threadId;
+  private String message;
   private String exceptionText;
 
-  public static AlertListenerMessage create(Object recipient, int msgLevel, Date msgDate,
-      String connectionName, String threadName, long tid, String msg, String exceptionText)
{
-    AlertListenerMessage m = new AlertListenerMessage();
-    m.setRecipient((InternalDistributedMember) recipient);
-    m.msgLevel = msgLevel;
-    m.msgDate = msgDate;
-    m.connectionName = connectionName;
-    if (m.connectionName == null) {
-      m.connectionName = "";
+  public static AlertListenerMessage create(DistributedMember recipient, int alertLevel,
Date date,
+      String connectionName, String threadName, long threadId, String message,
+      String exceptionText) {
+    AlertListenerMessage alertListenerMessage = new AlertListenerMessage();
+    alertListenerMessage.setRecipient((InternalDistributedMember) recipient);
+    alertListenerMessage.alertLevel = alertLevel;
+    alertListenerMessage.date = date;
+    alertListenerMessage.connectionName = connectionName;
+    if (alertListenerMessage.connectionName == null) {
+      alertListenerMessage.connectionName = "";
     }
-    m.threadName = threadName;
-    if (m.threadName == null) {
-      m.threadName = "";
+    alertListenerMessage.threadName = threadName;
+    if (alertListenerMessage.threadName == null) {
+      alertListenerMessage.threadName = "";
     }
-    m.tid = tid;
-    m.msg = msg;
-    if (m.msg == null) {
-      m.msg = "";
+    alertListenerMessage.threadId = threadId;
+    alertListenerMessage.message = message;
+    if (alertListenerMessage.message == null) {
+      alertListenerMessage.message = "";
     }
-    m.exceptionText = exceptionText;
-    if (m.exceptionText == null) {
-      m.exceptionText = "";
+    alertListenerMessage.exceptionText = exceptionText;
+    if (alertListenerMessage.exceptionText == null) {
+      alertListenerMessage.exceptionText = "";
     }
-    return m;
+    return alertListenerMessage;
   }
 
   @Override
   public void process(ClusterDistributionManager dm) {
     RemoteGfManagerAgent agent = dm.getAgent();
     if (agent != null) {
-      RemoteGemFireVM mgr = agent.getMemberById(this.getSender());
-      if (mgr == null)
+      RemoteGemFireVM manager = agent.getMemberById(getSender());
+      if (manager == null) {
         return;
-      Alert alert = new RemoteAlert(mgr, msgLevel, msgDate, connectionName, threadName, tid,
msg,
-          exceptionText, getSender());
+      }
+      Alert alert = new RemoteAlert(manager, alertLevel, date, connectionName, threadName,
threadId,
+          message, exceptionText, getSender());
       agent.callAlertListener(alert);
     } else {
-      /**
-       * Its assumed that its a managing node and it has to emit any alerts emitted to it.
+      /*
+       * The other recipient type is a JMX Manager which needs AlertDetails so that it can
send out
+       * JMX notifications for the alert.
        */
-      AlertDetails alertDetail = new AlertDetails(msgLevel, msgDate, connectionName, threadName,
-          tid, msg, exceptionText, getSender());
+      AlertDetails alertDetail = new AlertDetails(alertLevel, date, connectionName, threadName,
+          threadId, message, exceptionText, getSender());
       dm.getSystem().handleResourceEvent(ResourceEvent.SYSTEM_ALERT, alertDetail);
     }
-
   }
 
   @Override
@@ -96,6 +103,7 @@ public class AlertListenerMessage extends PooledDistributionMessage implements
A
     return true;
   }
 
+  @Override
   public int getDSFID() {
     return ALERT_LISTENER_MESSAGE;
   }
@@ -103,30 +111,29 @@ public class AlertListenerMessage extends PooledDistributionMessage
implements A
   @Override
   public void toData(DataOutput out) throws IOException {
     super.toData(out);
-    out.writeInt(msgLevel);
-    DataSerializer.writeObject(msgDate, out);
+    out.writeInt(alertLevel);
+    DataSerializer.writeObject(date, out);
     DataSerializer.writeString(connectionName, out);
     DataSerializer.writeString(threadName, out);
-    out.writeLong(tid);
-    DataSerializer.writeString(msg, out);
+    out.writeLong(threadId);
+    DataSerializer.writeString(message, out);
     DataSerializer.writeString(exceptionText, out);
   }
 
   @Override
   public void fromData(DataInput in) throws IOException, ClassNotFoundException {
     super.fromData(in);
-    this.msgLevel = in.readInt();
-    this.msgDate = (Date) DataSerializer.readObject(in);
-    this.connectionName = DataSerializer.readString(in);
-    this.threadName = DataSerializer.readString(in);
-    this.tid = in.readLong();
-    this.msg = DataSerializer.readString(in);
-    this.exceptionText = DataSerializer.readString(in);
+    alertLevel = in.readInt();
+    date = DataSerializer.readObject(in);
+    connectionName = DataSerializer.readString(in);
+    threadName = DataSerializer.readString(in);
+    threadId = in.readLong();
+    message = DataSerializer.readString(in);
+    exceptionText = DataSerializer.readString(in);
   }
 
   @Override
   public String toString() {
-    return "Alert \"" + this.msg + "\" level " + AlertLevel.forSeverity(this.msgLevel);
+    return "Alert \"" + message + "\" level " + AlertLevel.forSeverity(alertLevel);
   }
-
 }
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
b/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
index 6929e46..8b231ad 100755
--- a/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
@@ -56,6 +56,7 @@ import org.apache.geode.management.ManagementException;
  * Manager implementation which manages federated MBeans for the entire DistributedSystem
and
  * controls the JMX server endpoints for JMX clients to connect, such as an RMI server.
  *
+ * <p>
  * The FederatingManager is only appropriate for a peer or server in a GemFire distributed
system.
  *
  * @since GemFire 7.0
@@ -74,34 +75,26 @@ public class FederatingManager extends Manager {
   /**
    * Proxy factory is used to create , remove proxies
    */
-  protected MBeanProxyFactory proxyFactory;
-
-  private MBeanJMXAdapter jmxAdapter;
+  private MBeanProxyFactory proxyFactory;
 
   private MemberMessenger messenger;
 
-  private SystemManagementService service;
+  private final SystemManagementService service;
 
-  private AtomicReference<Exception> latestException = new AtomicReference<>(null);
+  private final AtomicReference<Exception> latestException = new AtomicReference<>(null);
 
-  /**
-   * @param jmxAdapter JMX Adapter
-   * @param repo Management resource repo
-   * @param system Internal Distributed System
-   * @param service SystemManagement Service
-   */
-  public FederatingManager(MBeanJMXAdapter jmxAdapter, ManagementResourceRepo repo,
-      InternalDistributedSystem system, SystemManagementService service, InternalCache cache)
{
+  FederatingManager(MBeanJMXAdapter jmxAdapter, ManagementResourceRepo repo,
+      InternalDistributedSystem system, SystemManagementService service,
+      InternalCache cache) {
     super(repo, system, cache);
     this.service = service;
-    this.proxyFactory = new MBeanProxyFactory(jmxAdapter, service);
-    this.jmxAdapter = jmxAdapter;
-    this.messenger = new MemberMessenger(jmxAdapter, repo, system);
+    proxyFactory = new MBeanProxyFactory(jmxAdapter, service);
+    messenger = new MemberMessenger(jmxAdapter, system);
   }
 
   @TestingOnly
   void setProxyFactory(MBeanProxyFactory newProxyFactory) {
-    this.proxyFactory = newProxyFactory;
+    proxyFactory = newProxyFactory;
   }
 
   /**
@@ -115,21 +108,19 @@ public class FederatingManager extends Manager {
         logger.debug("Starting the Federating Manager.... ");
       }
 
-      Runtime rt = Runtime.getRuntime();
-      this.pooledMembershipExecutor =
-          LoggingExecutors.newFixedThreadPool("FederatingManager", false, rt.availableProcessors());
+      pooledMembershipExecutor = LoggingExecutors.newFixedThreadPool("FederatingManager",
false,
+          Runtime.getRuntime().availableProcessors());
 
       running = true;
       startManagingActivity();
-
       messenger.broadcastManagerInfo();
-
     } catch (Exception e) {
       running = false;
       throw new ManagementException(e);
     }
   }
 
+  @Override
   public synchronized void stopManager() {
     // remove hidden management regions and federatedMBeans
     if (!running) {
@@ -143,17 +134,16 @@ public class FederatingManager extends Manager {
   }
 
   /**
-   * This method will be invoked whenever a member stops being a managing node. The exception
-   * Management exception has to be handled by the caller. *
+   * This method will be invoked whenever a member stops being a managing node. The
+   * {@code ManagementException} has to be handled by the caller.
    */
   private void stopManagingActivity() {
     try {
-      this.pooledMembershipExecutor.shutdownNow();
+      pooledMembershipExecutor.shutdownNow();
 
       for (DistributedMember distributedMember : repo.getMonitoringRegionMap().keySet())
{
         removeMemberArtifacts(distributedMember, false);
       }
-
     } catch (Exception e) {
       throw new ManagementException(e);
     }
@@ -168,6 +158,7 @@ public class FederatingManager extends Manager {
    * This method will be invoked from MembershipListener which is registered when the member
becomes
    * a Management node.
    *
+   * <p>
    * This method will delegate task to another thread and exit, so that it wont block the
membership
    * listener
    */
@@ -177,7 +168,7 @@ public class FederatingManager extends Manager {
       try {
         giiTask.call();
       } catch (Exception e) {
-        logger.warn("Error federating new member {}: {}", member.getId(), e.getMessage());
+        logger.warn("Error federating new member {}", member.getId(), e);
         latestException.set(e);
       }
     });
@@ -187,6 +178,7 @@ public class FederatingManager extends Manager {
    * This method will be invoked from MembershipListener which is registered when the member
becomes
    * a Management node.
    *
+   * <p>
    * This method will delegate task to another thread and exit, so that it wont block the
membership
    * listener
    */
@@ -198,53 +190,40 @@ public class FederatingManager extends Manager {
   private void executeTask(Runnable task) {
     try {
       pooledMembershipExecutor.execute(task);
-    } catch (RejectedExecutionException ex) {
+    } catch (RejectedExecutionException ignored) {
       // Ignore, we are getting shutdown
     }
   }
 
-  private class RemoveMemberTask implements Runnable {
-
-    private DistributedMember member;
-
-    boolean crashed;
-
-    protected RemoveMemberTask(DistributedMember member, boolean crashed) {
-      this.member = member;
-      this.crashed = crashed;
-    }
-
-    public void run() {
-      removeMemberArtifacts(member, crashed);
-    }
-  }
-
-  private DistributedMember removeMemberArtifacts(DistributedMember member, boolean crashed)
{
+  private void removeMemberArtifacts(DistributedMember member, boolean crashed) {
     Region<String, Object> proxyRegion = repo.getEntryFromMonitoringRegionMap(member);
-    Region<NotificationKey, Notification> notifRegion = repo.getEntryFromNotifRegionMap(member);
+    Region<NotificationKey, Notification> notificationRegion =
+        repo.getEntryFromNotifRegionMap(member);
 
-    if (proxyRegion == null && notifRegion == null) {
-      return member;
+    if (proxyRegion == null && notificationRegion == null) {
+      return;
     }
+
     repo.romoveEntryFromMonitoringRegionMap(member);
     repo.removeEntryFromNotifRegionMap(member);
-    // If cache is closed all the regions would have been
-    // destroyed implicitly
+
+    // If cache is closed all the regions would have been destroyed implicitly
     if (!cache.isClosed()) {
       proxyFactory.removeAllProxies(member, proxyRegion);
       proxyRegion.localDestroyRegion();
-      notifRegion.localDestroyRegion();
+      notificationRegion.localDestroyRegion();
     }
+
     if (!cache.getDistributedSystem().getDistributedMember().equals(member)) {
       service.memberDeparted((InternalDistributedMember) member, crashed);
     }
-    return member;
   }
 
   /**
    * This method will be invoked from MembershipListener which is registered when the member
becomes
    * a Management node.
    *
+   * <p>
    * this method will delegate task to another thread and exit, so that it wont block the
membership
    * listener
    */
@@ -258,11 +237,10 @@ public class FederatingManager extends Manager {
    * method will block for all GIIs to be completed But each GII is given a specific time
frame.
    * After that the task will be marked as cancelled.
    */
-  public void startManagingActivity() throws Exception {
-    final boolean isDebugEnabled = logger.isDebugEnabled();
+  private void startManagingActivity() {
+    boolean isDebugEnabled = logger.isDebugEnabled();
 
-    final List<Callable<DistributedMember>> giiTaskList = new ArrayList<>();
-    List<Future<DistributedMember>> futureTaskList;
+    List<Callable<DistributedMember>> giiTaskList = new ArrayList<>();
 
     for (DistributedMember member : cache.getDistributionManager()
         .getOtherDistributionManagerIds()) {
@@ -273,17 +251,13 @@ public class FederatingManager extends Manager {
       if (isDebugEnabled) {
         logger.debug("Management Resource creation started  : ");
       }
-      futureTaskList = pooledMembershipExecutor.invokeAll(giiTaskList);
+      List<Future<DistributedMember>> futureTaskList =
+          pooledMembershipExecutor.invokeAll(giiTaskList);
 
       for (Future<DistributedMember> futureTask : futureTaskList) {
-
-        String memberId = null;
         try {
-
           DistributedMember returnedMember = futureTask.get();
-          if (returnedMember != null) {
-            memberId = returnedMember.getId();
-          }
+          String memberId = returnedMember != null ? returnedMember.getId() : null;
 
           if (futureTask.isDone()) {
             if (isDebugEnabled) {
@@ -300,22 +274,20 @@ public class FederatingManager extends Manager {
           }
         } catch (ExecutionException e) {
           if (isDebugEnabled) {
-            logger.debug("ExecutionException during Management GII: {}", e.getMessage(),
e);
+            logger.debug("ExecutionException during Management GII", e);
           }
 
         } catch (CancellationException e) {
           if (isDebugEnabled) {
-            ManagementException mgEx = new ManagementException(e.fillInStackTrace());
-            logger.debug("InterruptedException while creating Monitoring resource with error
: {}",
-                mgEx.getMessage(), mgEx);
+            logger.debug("InterruptedException while creating Monitoring resource with error",
+                new ManagementException(e));
           }
         }
       }
     } catch (InterruptedException e) {
       if (isDebugEnabled) {
-        ManagementException mgEx = new ManagementException(e.fillInStackTrace());
-        logger.debug("InterruptedException while creating Monitoring resource with error
: ",
-            mgEx.getMessage(), mgEx);
+        logger.debug("InterruptedException while creating Monitoring resource with error",
+            new ManagementException(e));
       }
 
     } finally {
@@ -325,36 +297,62 @@ public class FederatingManager extends Manager {
     }
   }
 
+  private class RemoveMemberTask implements Runnable {
+
+    private final DistributedMember member;
+
+    boolean crashed;
+
+    RemoveMemberTask(DistributedMember member, boolean crashed) {
+      this.member = member;
+      this.crashed = crashed;
+    }
+
+    @Override
+    public void run() {
+      removeMemberArtifacts(member, crashed);
+    }
+  }
+
   /**
    * Actual task of doing the GII
    *
+   * <p>
    * It will perform the GII request which might originate from TransitionListener or Membership
    * Listener.
    *
-   * Managing Node side resources are created per member which is visible to this node
+   * <p>
+   * Managing Node side resources are created per member which is visible to this node:
    *
-   * 1)Management Region : its a Replicated NO_ACK region 2)Notification Region : its a Replicated
-   * Proxy NO_ACK region
+   * <pre>
+   * 1)Management Region : its a Replicated NO_ACK region
+   * 2)Notification Region : its a Replicated Proxy NO_ACK region
+   * </pre>
    *
-   * Listeners are added to the above two regions 1) ManagementCacheListener() 2)
-   * NotificationCacheListener
+   * <p>
+   * Listeners are added to the above two regions:
    *
+   * <pre>
+   * 1) ManagementCacheListener
+   * 2) NotificationCacheListener
+   * </pre>
+   *
+   * <p>
    * This task can be cancelled from the calling thread if a timeout happens. In that case
we have
    * to handle the thread interrupt
    */
   private class GIITask implements Callable<DistributedMember> {
 
-    private DistributedMember member;
+    private final DistributedMember member;
 
-    protected GIITask(DistributedMember member) {
+    GIITask(DistributedMember member) {
 
       this.member = member;
     }
 
+    @Override
     public DistributedMember call() {
       synchronized (member) {
-        Region<String, Object> proxyMonitoringRegion;
-        Region<NotificationKey, Notification> proxyNotificationRegion;
         String appender = MBeanJMXAdapter.getUniqueIDForMember(member);
         String monitoringRegionName = ManagementConstants.MONITORING_REGION + "_" + appender;
         String notificationRegionName = ManagementConstants.NOTIFICATION_REGION + "_" + appender;
@@ -370,81 +368,84 @@ public class FederatingManager extends Manager {
           if (!Thread.currentThread().isInterrupted()) {
 
             // as the regions will be internal regions
-            InternalRegionArguments internalArgs = new InternalRegionArguments();
-            internalArgs.setIsUsedForMetaRegion(true);
+            InternalRegionArguments internalRegionArguments = new InternalRegionArguments();
+            internalRegionArguments.setIsUsedForMetaRegion(true);
 
             // Create anonymous stats holder for Management Regions
-            final HasCachePerfStats monitoringRegionStats = new HasCachePerfStats() {
-              public CachePerfStats getCachePerfStats() {
-                return new CachePerfStats(cache.getDistributedSystem(), "managementRegionStats");
-              }
-            };
+            HasCachePerfStats monitoringRegionStats =
+                () -> new CachePerfStats(cache.getDistributedSystem(), "managementRegionStats");
 
-            internalArgs.setCachePerfStatsHolder(monitoringRegionStats);
+            internalRegionArguments.setCachePerfStatsHolder(monitoringRegionStats);
 
             // Monitoring region for member is created
-            AttributesFactory<String, Object> monitorAttrFactory = new AttributesFactory<>();
-            monitorAttrFactory.setScope(Scope.DISTRIBUTED_NO_ACK);
-            monitorAttrFactory.setDataPolicy(DataPolicy.REPLICATE);
-            monitorAttrFactory.setConcurrencyChecksEnabled(false);
-            ManagementCacheListener mgmtCacheListener = new ManagementCacheListener(proxyFactory);
-            monitorAttrFactory.addCacheListener(mgmtCacheListener);
+            AttributesFactory<String, Object> monitorAttributesFactory = new AttributesFactory<>();
+            monitorAttributesFactory.setScope(Scope.DISTRIBUTED_NO_ACK);
+            monitorAttributesFactory.setDataPolicy(DataPolicy.REPLICATE);
+            monitorAttributesFactory.setConcurrencyChecksEnabled(false);
+            ManagementCacheListener managementCacheListener =
+                new ManagementCacheListener(proxyFactory);
+            monitorAttributesFactory.addCacheListener(managementCacheListener);
 
-            RegionAttributes<String, Object> monitoringRegionAttrs = monitorAttrFactory.create();
+            RegionAttributes<String, Object> monitoringRegionAttrs =
+                monitorAttributesFactory.create();
 
             // Notification region for member is created
-            AttributesFactory<NotificationKey, Notification> notifAttrFactory =
+            AttributesFactory<NotificationKey, Notification> notificationAttributesFactory
=
                 new AttributesFactory<>();
-            notifAttrFactory.setScope(Scope.DISTRIBUTED_NO_ACK);
-            notifAttrFactory.setDataPolicy(DataPolicy.REPLICATE);
-            notifAttrFactory.setConcurrencyChecksEnabled(false);
+            notificationAttributesFactory.setScope(Scope.DISTRIBUTED_NO_ACK);
+            notificationAttributesFactory.setDataPolicy(DataPolicy.REPLICATE);
+            notificationAttributesFactory.setConcurrencyChecksEnabled(false);
 
             // Fix for issue #49638, evict the internal region _notificationRegion
-            notifAttrFactory.setEvictionAttributes(EvictionAttributes.createLRUEntryAttributes(
-                ManagementConstants.NOTIF_REGION_MAX_ENTRIES, EvictionAction.LOCAL_DESTROY));
+            notificationAttributesFactory
+                .setEvictionAttributes(EvictionAttributes.createLRUEntryAttributes(
+                    ManagementConstants.NOTIF_REGION_MAX_ENTRIES, EvictionAction.LOCAL_DESTROY));
 
             NotificationCacheListener notifListener = new NotificationCacheListener(proxyFactory);
-            notifAttrFactory.addCacheListener(notifListener);
+            notificationAttributesFactory.addCacheListener(notifListener);
 
             RegionAttributes<NotificationKey, Notification> notifRegionAttrs =
-                notifAttrFactory.create();
+                notificationAttributesFactory.create();
 
-            boolean proxyMonitoringRegionCreated = false;
-            boolean proxyNotifRegionCreated = false;
+            boolean proxyMonitoringRegionCreated;
 
+            Region<String, Object> proxyMonitoringRegion;
             try {
               if (!running) {
                 return null;
               }
               proxyMonitoringRegion =
-                  cache.createVMRegion(monitoringRegionName, monitoringRegionAttrs, internalArgs);
+                  cache.createVMRegion(monitoringRegionName, monitoringRegionAttrs,
+                      internalRegionArguments);
               proxyMonitoringRegionCreated = true;
 
             } catch (TimeoutException | RegionExistsException | IOException
                 | ClassNotFoundException e) {
               if (logger.isDebugEnabled()) {
-                logger.debug("Error During Internal Region creation {}", e.getMessage(),
e);
+                logger.debug("Error During Internal Region creation", e);
               }
               throw new ManagementException(e);
             }
 
+            boolean proxyNotificationRegionCreated = false;
+            Region<NotificationKey, Notification> proxyNotificationRegion;
             try {
               if (!running) {
                 return null;
               }
               proxyNotificationRegion =
-                  cache.createVMRegion(notificationRegionName, notifRegionAttrs, internalArgs);
-              proxyNotifRegionCreated = true;
+                  cache.createVMRegion(notificationRegionName, notifRegionAttrs,
+                      internalRegionArguments);
+              proxyNotificationRegionCreated = true;
             } catch (TimeoutException | RegionExistsException | IOException
                 | ClassNotFoundException e) {
               if (logger.isDebugEnabled()) {
-                logger.debug("Error During Internal Region creation {}", e.getMessage(),
e);
+                logger.debug("Error During Internal Region creation", e);
               }
               throw new ManagementException(e);
             } finally {
-              if (!proxyNotifRegionCreated && proxyMonitoringRegionCreated) {
-                // Destroy the proxy region if proxy notification
-                // region is not created
+              if (!proxyNotificationRegionCreated && proxyMonitoringRegionCreated)
{
+                // Destroy the proxy region if proxy notification region is not created
                 proxyMonitoringRegion.localDestroyRegion();
               }
             }
@@ -466,11 +467,11 @@ public class FederatingManager extends Manager {
               }
               proxyFactory.createAllProxies(member, proxyMonitoringRegion);
 
-              mgmtCacheListener.markReady();
+              managementCacheListener.markReady();
               notifListener.markReady();
             } catch (Exception e) {
               if (logger.isDebugEnabled()) {
-                logger.debug("Error During GII Proxy creation {}", e.getMessage(), e);
+                logger.debug("Error During GII Proxy creation", e);
               }
 
               throw new ManagementException(e);
@@ -501,36 +502,38 @@ public class FederatingManager extends Manager {
   }
 
   /**
-   * This will return the last updated time of the proxyMBean
+   * This will return the last updated time of the proxyMBean.
+   *
+   * @param objectName {@link ObjectName} of the MBean
    *
-   * @param objectName {@link javax.management.ObjectName} of the MBean
    * @return last updated time of the proxy
    */
-  public long getLastUpdateTime(ObjectName objectName) {
+  long getLastUpdateTime(ObjectName objectName) {
     return proxyFactory.getLastUpdateTime(objectName);
   }
 
   /**
-   * Find a particular proxy instance for a {@link javax.management.ObjectName} ,
-   * {@link org.apache.geode.distributed.DistributedMember} and interface class If the proxy
-   * interface does not implement the given interface class a {@link java.lang.ClassCastException}.
-   * will be thrown
+   * Find a particular proxy instance for a {@link ObjectName}, {@link DistributedMember}
and
+   * interface class If the proxy interface does not implement the given interface class
a
+   * {@link ClassCastException} will be thrown
    *
-   * @param objectName {@link javax.management.ObjectName} of the MBean
+   * @param objectName {@link ObjectName} of the MBean
    * @param interfaceClass interface class implemented by proxy
+   *
    * @return an instance of proxy exposing the given interface
    */
-  public <T> T findProxy(ObjectName objectName, Class<T> interfaceClass) {
+  <T> T findProxy(ObjectName objectName, Class<T> interfaceClass) {
     return proxyFactory.findProxy(objectName, interfaceClass);
   }
 
   /**
-   * Find a set of proxies given a {@link org.apache.geode.distributed.DistributedMember}
+   * Find a set of proxies given a {@link DistributedMember}.
+   *
+   * @param member {@link DistributedMember}
    *
-   * @param member {@link org.apache.geode.distributed.DistributedMember}
-   * @return a set of {@link javax.management.ObjectName}
+   * @return a set of {@link ObjectName}
    */
-  public Set<ObjectName> findAllProxies(DistributedMember member) {
+  Set<ObjectName> findAllProxies(DistributedMember member) {
     return proxyFactory.findAllProxies(member);
   }
 
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/ManagerStartupMessage.java
b/geode-core/src/main/java/org/apache/geode/management/internal/ManagerStartupMessage.java
index 656bdd7..ec4457c 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/ManagerStartupMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/ManagerStartupMessage.java
@@ -23,28 +23,33 @@ import org.apache.geode.distributed.internal.PooledDistributionMessage;
 import org.apache.geode.internal.admin.Alert;
 import org.apache.geode.internal.logging.log4j.AlertAppender;
 
+/**
+ * Sent by JMX manager to all other members to notify them that it has started.
+ */
 public class ManagerStartupMessage extends PooledDistributionMessage {
-  // instance variables
-  int alertLevel;
 
-  public static ManagerStartupMessage create(int level) {
-    ManagerStartupMessage m = new ManagerStartupMessage();
-    m.setLevel(level);
-    return m;
+  private int alertLevel;
+
+  public static ManagerStartupMessage create(int alertLevel) {
+    return new ManagerStartupMessage(alertLevel);
+  }
+
+  public ManagerStartupMessage() {
+    // required for DataSerializableFixedID
   }
 
-  public void setLevel(int alertLevel) {
+  private ManagerStartupMessage(int alertLevel) {
     this.alertLevel = alertLevel;
   }
 
   @Override
   public void process(ClusterDistributionManager dm) {
-
-    if (this.alertLevel != Alert.OFF) {
-      AlertAppender.getInstance().addAlertListener(this.getSender(), this.alertLevel);
+    if (alertLevel != Alert.OFF) {
+      AlertAppender.getInstance().addAlertListener(getSender(), alertLevel);
     }
   }
 
+  @Override
   public int getDSFID() {
     return MANAGER_STARTUP_MESSAGE;
   }
@@ -52,19 +57,17 @@ public class ManagerStartupMessage extends PooledDistributionMessage {
   @Override
   public void toData(DataOutput out) throws IOException {
     super.toData(out);
-    out.writeInt(this.alertLevel);
+    out.writeInt(alertLevel);
   }
 
   @Override
   public void fromData(DataInput in) throws IOException, ClassNotFoundException {
     super.fromData(in);
-    this.alertLevel = in.readInt();
+    alertLevel = in.readInt();
   }
 
   @Override
   public String toString() {
-    return "ManagerStartupMessage from " + this.getSender() + " level=" + alertLevel;
+    return "ManagerStartupMessage from " + getSender() + " alertLevel=" + alertLevel;
   }
-
-
 }
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/MemberMessenger.java
b/geode-core/src/main/java/org/apache/geode/management/internal/MemberMessenger.java
index 0f84bf9..30a73fd 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/MemberMessenger.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/MemberMessenger.java
@@ -12,7 +12,6 @@
  * or implied. See the License for the specific language governing permissions and limitations
under
  * the License.
  */
-
 package org.apache.geode.management.internal;
 
 import java.util.Set;
@@ -27,43 +26,77 @@ import org.apache.geode.internal.admin.remote.AlertLevelChangeMessage;
 import org.apache.geode.internal.logging.log4j.LogLevel;
 
 /**
- * This class will act as a messenger from manager to members for various operations.
- *
- * To start with its is designed to Manager start stop details, change alert level.
- *
- *
- *
+ * Handles messaging from manager to members for various operations. It sends two types of
messages:
+ * {@link ManagerStartupMessage} and {@link AlertLevelChangeMessage}
  */
 public class MemberMessenger {
 
-  private MBeanJMXAdapter jmxAdapter;
-  private ManagementResourceRepo repo;
-  private InternalDistributedSystem system;
+  private final MBeanJMXAdapter jmxAdapter;
+  private final InternalDistributedSystem system;
 
-  public MemberMessenger(MBeanJMXAdapter jmxAdapter, ManagementResourceRepo repo,
-      InternalDistributedSystem system) {
+  MemberMessenger(MBeanJMXAdapter jmxAdapter, InternalDistributedSystem system) {
     this.jmxAdapter = jmxAdapter;
-    this.repo = repo;
     this.system = system;
-
   }
 
-  public void sendManagerInfo(DistributedMember receiver) {
-
+  /**
+   * <pre>
+   * Code path 1:  {@link FederatingManager#startManager()} ->
+   *               {@link FederatingManager#startManagingActivity()} ->
+   *               {@link FederatingManager.GIITask#call()} ->
+   *               {@code sendManagerInfo}
+   * Recipient(s): {@link DistributionManager#getOtherDistributionManagerIds()}
+   * When:         starting the manager
+   * </pre>
+   *
+   * <p>
+   *
+   * <pre>
+   * Code path 2:  {@link FederatingManager#addMember(DistributedMember)} ->
+   *               {@link FederatingManager.GIITask#call()} ->
+   *               {@code sendManagerInfo}
+   * Recipient(s): one new member from membership listener
+   * When:         new member joins
+   * </pre>
+   *
+   * <p>
+   * This path does <b>NOT</b> process {@link ManagerStartupMessage} locally
so we do not add an
+   * AlertListener for local Alerts.
+   */
+  void sendManagerInfo(DistributedMember receiver) {
     String levelName = jmxAdapter.getDistributedSystemMXBean().getAlertLevel();
     int alertCode = LogLevel.getLogWriterLevel(levelName);
+
     ManagerStartupMessage msg = ManagerStartupMessage.create(alertCode);
     msg.setRecipient((InternalDistributedMember) receiver);
-    sendAsync(msg);
 
+    sendAsync(msg);
   }
 
-  public void broadcastManagerInfo() {
+  /**
+   * <pre>
+   * Code path:    {@link FederatingManager#startManager()} ->
+   *               {@code sendManagerInfo}
+   * Recipient(s): {@link DistributionManager#getAllOtherMembers()}
+   * When:         starting the manager
+   * </pre>
+   *
+   * <p>
+   * This call seems to be redundant with {@link #sendManagerInfo(DistributedMember)} except
that
+   * here we do not switch context to another thread and we also register a local AlertListener.
+   *
+   * <p>
+   * After sending to remote members, this call processes the {@link ManagerStartupMessage}
locally
+   * which would add an AlertListener for local Alerts. But local Alerts don't seem to work
+   * (GEODE-5923).
+   */
+  void broadcastManagerInfo() {
     Set<InternalDistributedMember> otherMemberSet =
         system.getDistributionManager().getAllOtherMembers();
 
     String levelName = jmxAdapter.getDistributedSystemMXBean().getAlertLevel();
     int alertCode = LogLevel.getLogWriterLevel(levelName);
+
     ManagerStartupMessage msg = ManagerStartupMessage.create(alertCode);
     if (otherMemberSet != null && otherMemberSet.size() > 0) {
       msg.setRecipients(otherMemberSet);
@@ -75,14 +108,15 @@ public class MemberMessenger {
     if (dm instanceof ClusterDistributionManager) {
       msg.process((ClusterDistributionManager) system.getDistributionManager());
     }
-
-
   }
 
   /**
-   * Sends a message and does not wait for a response
+   * Sends a message and does not wait for a response.
+   *
+   * <p>
+   * Actually, it's the message implementation that determines if it's sync or async, not
this call.
    */
-  void sendAsync(DistributionMessage msg) {
+  private void sendAsync(DistributionMessage msg) {
     if (system != null) {
       system.getDistributionManager().putOutgoing(msg);
     }


Mime
View raw message