/*
 *
 * 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.tools.pulse.internal.data;

import java.io.IOException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.ResourceBundle;
import java.util.Set;
import java.util.TimeZone;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.management.remote.JMXConnector;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.commons.collections.buffer.CircularFifoBuffer;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * Class Cluster This class is the Data Model for the data used for the Pulse Web UI.
 *
 * @since GemFire version 7.0.Beta 2012-09-23
 */
public class Cluster extends Thread {
  private static final int POLL_INTERVAL = 5000;
  public static final int MAX_SAMPLE_SIZE = 180;
  public static final int ALERTS_MAX_SIZE = 1000;
  public static final int PAGE_ALERTS_MAX_SIZE = 100;

  private static final Logger logger = LogManager.getLogger();
  private Repository repository;
  private ResourceBundle resourceBundle;

  private String jmxUserName;
  private String serverName;
  private String port;
  private JMXConnector jmxConnector;
  private int stale = 0;
  private double loadPerSec;
  private CountDownLatch clusterHasBeenInitialized;


  // start: fields defined in System MBean
  private IClusterUpdater updater = null;
  private DataBrowser dataBrowser = null;
  private int memberCount;
  private long clientConnectionCount;
  private int locatorCount;
  private int totalRegionCount;
  private long totalHeapSize = 0L;
  private long totalRegionEntryCount;
  private int currentQueryCount;
  private long totalBytesOnDisk;
  private double diskReadsRate;
  private double diskWritesRate;
  private double writePerSec;
  private double readPerSec;
  private double queriesPerSec;
  private int avgDiskStorage;
  private int avgDiskWritesRate;
  private int runningFunctionCount;
  private long registeredCQCount;
  private int subscriptionCount;
  private int serverCount;
  private int txnCommittedCount;
  private int txnRollbackCount;
  private long usedHeapSize = 0L;
  private long garbageCollectionCount = 0L;
  private int clusterId;
  private int notificationPageNumber = 1;
  private volatile boolean connectedFlag;
  private String connectionErrorMsg = "";

  private Set<String> deletedMembers = new HashSet<>();

  private Map<String, List<Cluster.Member>> physicalToMember = new HashMap<>();

  private Map<String, Cluster.Member> membersHMap = new HashMap<>();

  private Set<String> deletedRegions = new HashSet<>();

  private Map<String, Cluster.Region> clusterRegionMap = new ConcurrentHashMap<>();

  private final List<Cluster.Alert> alertsList = new ArrayList<>();

  private CircularFifoBuffer totalBytesOnDiskTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private CircularFifoBuffer throughoutWritesTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private CircularFifoBuffer throughoutReadsTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private CircularFifoBuffer writePerSecTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private CircularFifoBuffer readPerSecTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private CircularFifoBuffer queriesPerSecTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private CircularFifoBuffer memoryUsageTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private CircularFifoBuffer garbageCollectionTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
  private long previousJVMPauseCount = 0L;

  private final HashMap<String, Boolean> wanInformation = new HashMap<>();
  private Map<String, Cluster.Statement> clusterStatementMap = new ConcurrentHashMap<>();

  public static final int CLUSTER_STAT_TOTAL_BYTES_ON_DISK = 0;
  public static final int CLUSTER_STAT_THROUGHPUT_WRITES = 1;
  public static final int CLUSTER_STAT_THROUGHPUT_READS = 2;
  public static final int CLUSTER_STAT_WRITES_PER_SECOND = 3;
  public static final int CLUSTER_STAT_READ_PER_SECOND = 4;
  public static final int CLUSTER_STAT_QUERIES_PER_SECOND = 5;
  public static final int CLUSTER_STAT_MEMORY_USAGE = 6;
  public static final int CLUSTER_STAT_GARBAGE_COLLECTION = 7;

  // end: fields defined in System MBean

  // used for updating member's client data
  public static long LAST_UPDATE_TIME = 0;

  private boolean stopUpdates = false;

  public Object[] getStatisticTrend(int trendId) {

    Object[] returnArray = null;
    switch (trendId) {
      case CLUSTER_STAT_TOTAL_BYTES_ON_DISK:
        synchronized (totalBytesOnDiskTrend) {
          returnArray = totalBytesOnDiskTrend.toArray();
        }

        break;

      case CLUSTER_STAT_THROUGHPUT_READS:
        synchronized (throughoutReadsTrend) {
          returnArray = throughoutReadsTrend.toArray();
        }
        break;

      case CLUSTER_STAT_THROUGHPUT_WRITES:
        synchronized (throughoutWritesTrend) {
          returnArray = throughoutWritesTrend.toArray();
        }
        break;

      case CLUSTER_STAT_WRITES_PER_SECOND:
        synchronized (writePerSecTrend) {
          returnArray = writePerSecTrend.toArray();
        }
        break;

      case CLUSTER_STAT_READ_PER_SECOND:
        synchronized (readPerSecTrend) {
          returnArray = readPerSecTrend.toArray();
        }
        break;

      case CLUSTER_STAT_QUERIES_PER_SECOND:
        synchronized (queriesPerSecTrend) {
          returnArray = queriesPerSecTrend.toArray();
        }
        break;

      case CLUSTER_STAT_MEMORY_USAGE:
        synchronized (memoryUsageTrend) {
          returnArray = memoryUsageTrend.toArray();
        }
        break;

      case CLUSTER_STAT_GARBAGE_COLLECTION:
        synchronized (garbageCollectionTrend) {
          returnArray = garbageCollectionTrend.toArray();
        }
        break;
    }

    return returnArray;
  }

  /**
   * Member Inner Class
   *
   *
   */
  public static class Member {

    // start: fields defined in MBean
    private String gemfireVersion;
    private boolean manager;
    private int totalRegionCount;
    private String host;
    private String hostnameForClients;
    private String bindAddress;
    private long currentHeapSize;
    private long maxHeapSize;
    private int avgHeapUsage;
    private long OffHeapFreeSize;
    private long OffHeapUsedSize;
    private long totalBytesOnDisk;
    private String memberPort;

    private double cpuUsage = 0.0d;
    private double hostCpuUsage = 0.0d;
    private long uptime;
    private String name;
    private double getsRate;
    private double putsRate;
    private boolean isCache;
    private boolean isGateway;
    private boolean isLocator;
    private boolean isServer;
    private double loadAverage;
    private int numThreads;
    private long totalFileDescriptorOpen;
    private long garbageCollectionCount = 0L;
    private double throughputWrites;
    private double throughputReads;
    private long totalDiskUsage;
    private String queueBacklog;
    private String id;
    private long numSqlfireClients = 0;

    private List<String> serverGroups = new ArrayList<>();
    private List<String> redundancyZones = new ArrayList<>();

    private CircularFifoBuffer cpuUsageSamples = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private CircularFifoBuffer heapUsageSamples = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private HashMap<String, Cluster.Region> memberRegions = new HashMap<>();
    private HashMap<String, Cluster.Client> memberClientsHMap =
        new HashMap<>();
    private CircularFifoBuffer totalBytesOnDiskSamples = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private CircularFifoBuffer getsPerSecond = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private CircularFifoBuffer putsPerSecond = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private CircularFifoBuffer throughputWritesTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private CircularFifoBuffer throughputReadsTrend = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private CircularFifoBuffer garbageCollectionSamples = new CircularFifoBuffer(MAX_SAMPLE_SIZE);
    private long previousJVMPauseCount = 0L;

    private Cluster.GatewayReceiver gatewayReceiver = null;
    private List<Cluster.GatewaySender> gatewaySenderList = new ArrayList<>();
    private List<Cluster.AsyncEventQueue> asyncEventQueueList =
        new ArrayList<>();
    // end: fields defined in MBean

    public static final int MEMBER_STAT_GARBAGE_COLLECTION = 0;
    public static final int MEMBER_STAT_HEAP_USAGE_SAMPLE = 1;
    public static final int MEMBER_STAT_CPU_USAGE_SAMPLE = 2;
    public static final int MEMBER_STAT_GETS_PER_SECOND = 3;
    public static final int MEMBER_STAT_PUTS_PER_SECOND = 4;
    public static final int MEMBER_STAT_THROUGHPUT_WRITES = 5;
    public static final int MEMBER_STAT_THROUGHPUT_READS = 6;

    public Cluster.Region[] getMemberRegionsList() {
      Cluster.Region[] memberReg;
      synchronized (memberRegions) {
        memberReg = new Cluster.Region[memberRegions.size()];
        memberReg = memberRegions.values().toArray(memberReg);
      }

      return memberReg;
    }

    public Cluster.Client[] getMemberClients() {
      Cluster.Client[] memberClients;
      synchronized (memberClientsHMap) {
        memberClients = new Cluster.Client[memberClientsHMap.size()];
        memberClients = memberClientsHMap.values().toArray(memberClients);
      }

      return memberClients;
    }

    public Cluster.GatewaySender[] getMemberGatewaySenders() {
      Cluster.GatewaySender[] memberGWS;
      synchronized (gatewaySenderList) {
        memberGWS = new Cluster.GatewaySender[gatewaySenderList.size()];
        memberGWS = gatewaySenderList.toArray(memberGWS);
      }
      return memberGWS;
    }

    public Cluster.AsyncEventQueue[] getMemberAsyncEventQueueList() {
      Cluster.AsyncEventQueue[] memberAEQ;
      synchronized (asyncEventQueueList) {
        memberAEQ = new Cluster.AsyncEventQueue[asyncEventQueueList.size()];
        memberAEQ = asyncEventQueueList.toArray(memberAEQ);
      }
      return memberAEQ;
    }

    public Object[] getMemberStatisticTrend(int trendId) {
      Object[] returnArray = null;
      switch (trendId) {
        case MEMBER_STAT_GARBAGE_COLLECTION:
          synchronized (garbageCollectionSamples) {
            returnArray = garbageCollectionSamples.toArray();
          }

          break;

        case MEMBER_STAT_HEAP_USAGE_SAMPLE:
          synchronized (heapUsageSamples) {
            returnArray = heapUsageSamples.toArray();
          }
          break;

        case MEMBER_STAT_CPU_USAGE_SAMPLE:
          synchronized (cpuUsageSamples) {
            returnArray = cpuUsageSamples.toArray();
          }
          break;

        case MEMBER_STAT_GETS_PER_SECOND:
          synchronized (getsPerSecond) {
            returnArray = getsPerSecond.toArray();
          }
          break;

        case MEMBER_STAT_PUTS_PER_SECOND:
          synchronized (putsPerSecond) {
            returnArray = putsPerSecond.toArray();
          }
          break;

        case MEMBER_STAT_THROUGHPUT_WRITES:
          synchronized (throughputWritesTrend) {
            returnArray = throughputWritesTrend.toArray();
          }
          break;

        case MEMBER_STAT_THROUGHPUT_READS:
          synchronized (throughputReadsTrend) {
            returnArray = throughputReadsTrend.toArray();
          }
          break;
      }

      return returnArray;
    }

    public String getGemfireVersion() {
      return gemfireVersion;
    }

    public void setGemfireVersion(String gemfireVersion) {
      this.gemfireVersion = gemfireVersion;
    }

    public String getMemberPort() {
      return memberPort;
    }

    public void setMemberPort(String memberPort) {
      this.memberPort = memberPort;
    }

    public double getThroughputWrites() {
      return throughputWrites;
    }

    public void setThroughputWrites(double throughputWrites) {
      this.throughputWrites = throughputWrites;
    }

    public double getThroughputReads() {
      return throughputReads;
    }

    public void setThroughputReads(double throughputReads) {
      this.throughputReads = throughputReads;
    }

    public long getTotalDiskUsage() {
      return totalDiskUsage;
    }

    public void setTotalDiskUsage(long totalDiskUsage) {
      this.totalDiskUsage = totalDiskUsage;
    }

    public String getId() {
      return id;
    }

    public String getName() {
      return name;
    }

    public double getLoadAverage() {
      return loadAverage;
    }

    public void setLoadAverage(Double loadAverage) {
      this.loadAverage = loadAverage;
    }

    public String getHost() {
      return host;
    }

    public String getHostnameForClients() {
      if (StringUtils.isNotBlank(hostnameForClients)) {
        return hostnameForClients;
      } else if (StringUtils.isNotBlank(bindAddress)) {
        return bindAddress;
      }
      return null;
    }

    public long getUptime() {
      return uptime;
    }

    public String getQueueBacklog() {
      return queueBacklog;
    }

    public HashMap<String, Cluster.Region> getMemberRegions() {
      return memberRegions;
    }

    public void setMemberRegions(HashMap<String, Cluster.Region> memberRegions) {
      this.memberRegions = memberRegions;
    }

    public long getCurrentHeapSize() {
      return currentHeapSize;
    }

    public void setCurrentHeapSize(long currentHeapSize) {
      this.currentHeapSize = currentHeapSize;
    }

    public long getMaxHeapSize() {
      return maxHeapSize;
    }

    public void setMaxHeapSize(long maxHeapSize) {
      this.maxHeapSize = maxHeapSize;
    }

    public boolean isManager() {
      return manager;
    }

    public void setManager(boolean manager) {
      this.manager = manager;
    }

    public int getAvgHeapUsage() {
      return avgHeapUsage;
    }

    public void setAvgHeapUsage(int avgHeapUsage) {
      this.avgHeapUsage = avgHeapUsage;
    }

    public long getOffHeapFreeSize() {
      return OffHeapFreeSize;
    }

    public void setOffHeapFreeSize(long offHeapFreeSize) {
      OffHeapFreeSize = offHeapFreeSize;
    }

    public long getOffHeapUsedSize() {
      return OffHeapUsedSize;
    }

    public void setOffHeapUsedSize(long offHeapUsedSize) {
      OffHeapUsedSize = offHeapUsedSize;
    }

    public void setId(String id) {
      this.id = id;
    }

    public void setName(String name) {
      this.name = name;
    }

    public void setHost(String host) {
      this.host = host;
    }

    public void setHostnameForClients(String hostnameForClients) {
      this.hostnameForClients = hostnameForClients;
    }

    public void setBindAddress(String bindAddress) {
      this.bindAddress = bindAddress;
    }

    public void setUptime(long uptime) {
      this.uptime = uptime;
    }

    public void setQueueBacklog(String queueBacklog) {
      this.queueBacklog = queueBacklog;
    }

    public int getTotalRegionCount() {
      return totalRegionCount;
    }

    public void setTotalRegionCount(int totalRegionCount) {
      this.totalRegionCount = totalRegionCount;
    }

    public long getTotalBytesOnDisk() {
      return totalBytesOnDisk;
    }

    public void setTotalBytesOnDisk(long totalBytesOnDisk) {
      this.totalBytesOnDisk = totalBytesOnDisk;
    }

    public double getCpuUsage() {
      return cpuUsage;
    }

    public void setCpuUsage(double cpuUsage) {
      this.cpuUsage = cpuUsage;
    }

    public double getHostCpuUsage() {
      return hostCpuUsage;
    }

    public void setHostCpuUsage(double hostCpuUsage) {
      this.hostCpuUsage = hostCpuUsage;
    }

    public double getGetsRate() {
      return getsRate;
    }

    public void setGetsRate(double getsRate) {
      this.getsRate = getsRate;
    }

    public double getPutsRate() {
      return putsRate;
    }

    public void setPutsRate(double putsRate) {
      this.putsRate = putsRate;
    }

    public HashMap<String, Cluster.Client> getMemberClientsHMap() {
      return memberClientsHMap;
    }

    public void setMemberClientsHMap(HashMap<String, Cluster.Client> memberClientsHMap) {
      this.memberClientsHMap = memberClientsHMap;
    }

    public boolean isCache() {
      return isCache;
    }

    public void setCache(boolean isCache) {
      this.isCache = isCache;
    }

    public boolean isGateway() {
      return isGateway;
    }

    public void setGateway(boolean isGateway) {
      this.isGateway = isGateway;
    }

    public int getNumThreads() {
      return numThreads;
    }

    public void setNumThreads(int numThreads) {
      this.numThreads = numThreads;
    }

    public long getTotalFileDescriptorOpen() {
      return totalFileDescriptorOpen;
    }

    public void setTotalFileDescriptorOpen(long totalFileDescriptorOpen) {
      this.totalFileDescriptorOpen = totalFileDescriptorOpen;
    }

    public long getGarbageCollectionCount() {
      return garbageCollectionCount;
    }

    public void setGarbageCollectionCount(long garbageCollectionCount) {
      this.garbageCollectionCount = garbageCollectionCount;
    }

    public boolean isLocator() {
      return isLocator;
    }

    public void setLocator(boolean isLocator) {
      this.isLocator = isLocator;
    }

    public Cluster.GatewayReceiver getGatewayReceiver() {
      return gatewayReceiver;
    }

    public void setGatewayReceiver(Cluster.GatewayReceiver gatewayReceiver) {
      this.gatewayReceiver = gatewayReceiver;
    }

    public List<Cluster.GatewaySender> getGatewaySenderList() {
      return gatewaySenderList;
    }

    public void setGatewaySenderList(List<Cluster.GatewaySender> gatewaySenderList) {
      this.gatewaySenderList = gatewaySenderList;
    }

    public List<Cluster.AsyncEventQueue> getAsyncEventQueueList() {
      return asyncEventQueueList;
    }

    public void setAsyncEventQueueList(List<Cluster.AsyncEventQueue> asyncEventQueueList) {
      this.asyncEventQueueList = asyncEventQueueList;
    }

    public boolean isServer() {
      return isServer;
    }

    public void setServer(boolean isServer) {
      this.isServer = isServer;
    }

    public List<String> getServerGroups() {
      return serverGroups;
    }

    public void setServerGroups(List<String> serverGroups) {
      this.serverGroups = serverGroups;
    }

    public List<String> getRedundancyZones() {
      return redundancyZones;
    }

    public void setRedundancyZones(List<String> redundancyZones) {
      this.redundancyZones = redundancyZones;
    }

    public CircularFifoBuffer getCpuUsageSamples() {
      return cpuUsageSamples;
    }

    public void setCpuUsageSamples(CircularFifoBuffer cpuUsageSamples) {
      this.cpuUsageSamples = cpuUsageSamples;
    }

    public CircularFifoBuffer getHeapUsageSamples() {
      return heapUsageSamples;
    }

    public void setHeapUsageSamples(CircularFifoBuffer heapUsageSamples) {
      this.heapUsageSamples = heapUsageSamples;
    }

    public CircularFifoBuffer getTotalBytesOnDiskSamples() {
      return totalBytesOnDiskSamples;
    }

    public void setTotalBytesOnDiskSamples(CircularFifoBuffer totalBytesOnDiskSamples) {
      this.totalBytesOnDiskSamples = totalBytesOnDiskSamples;
    }

    public CircularFifoBuffer getGetsPerSecond() {
      return getsPerSecond;
    }

    public void setGetsPerSecond(CircularFifoBuffer getsPerSecond) {
      this.getsPerSecond = getsPerSecond;
    }

    public CircularFifoBuffer getPutsPerSecond() {
      return putsPerSecond;
    }

    public void setPutsPerSecond(CircularFifoBuffer putsPerSecond) {
      this.putsPerSecond = putsPerSecond;
    }

    public CircularFifoBuffer getThroughputWritesTrend() {
      return throughputWritesTrend;
    }

    public void setThroughputWritesTrend(CircularFifoBuffer throughputWritesTrend) {
      this.throughputWritesTrend = throughputWritesTrend;
    }

    public CircularFifoBuffer getThroughputReadsTrend() {
      return throughputReadsTrend;
    }

    public void setThroughputReadsTrend(CircularFifoBuffer throughputReadsTrend) {
      this.throughputReadsTrend = throughputReadsTrend;
    }

    public CircularFifoBuffer getGarbageCollectionSamples() {
      return garbageCollectionSamples;
    }

    public void setGarbageCollectionSamples(CircularFifoBuffer garbageCollectionSamples) {
      this.garbageCollectionSamples = garbageCollectionSamples;
    }

    public long getPreviousJVMPauseCount() {
      return previousJVMPauseCount;
    }

    public void setPreviousJVMPauseCount(long previousJVMPauseCount) {
      this.previousJVMPauseCount = previousJVMPauseCount;
    }

    public long getNumSqlfireClients() {
      return numSqlfireClients;
    }

    public void setNumSqlfireClients(long numSqlfireClients) {
      this.numSqlfireClients = numSqlfireClients;
    }

    public void updateMemberClientsHMap(HashMap<String, Cluster.Client> memberClientsHM) {
      if (Cluster.LAST_UPDATE_TIME == 0) {
        Cluster.LAST_UPDATE_TIME = System.nanoTime();
      }

      long systemNanoTime = System.nanoTime();

      for (Map.Entry<String, Cluster.Client> entry : memberClientsHM.entrySet()) {
        String clientId = entry.getKey();
        Cluster.Client client = entry.getValue();

        if (memberClientsHMap.get(clientId) != null) {
          Client existingClient = memberClientsHMap.get(clientId);
          Client updatedClient = memberClientsHM.get(clientId);

          existingClient.setConnected(updatedClient.isConnected());
          existingClient.setGets(updatedClient.getGets());
          existingClient.setPuts(updatedClient.getPuts());
          existingClient.setCpus(updatedClient.getCpus());
          existingClient.setQueueSize(updatedClient.getQueueSize());
          existingClient.setStatus(updatedClient.getStatus());
          existingClient.setThreads(updatedClient.getThreads());
          existingClient.setClientCQCount(updatedClient.getClientCQCount());
          existingClient.setSubscriptionEnabled(updatedClient.isSubscriptionEnabled());
          long elapsedTime = updatedClient.getUptime() - existingClient.getUptime();
          existingClient.setUptime(updatedClient.getUptime());

          // set cpu usage
          long currCPUTime = updatedClient.getProcessCpuTime();
          long lastCPUTime = existingClient.getProcessCpuTime();

          double newCPUTime = 0;
          if (elapsedTime > 0) {
            newCPUTime = (double) (((currCPUTime - lastCPUTime) / elapsedTime) / 1_000_000_000L);
          }

          double newCPUUsage = 0;
          int availableCpus = updatedClient.getCpus();
          if (availableCpus > 0) {
            newCPUUsage = newCPUTime / availableCpus;
          }

          existingClient.setCpuUsage(newCPUUsage);
          existingClient.setProcessCpuTime(currCPUTime);
        } else {
          // Add client to clients list
          memberClientsHMap.put(clientId, client);
        }
      }

      // Remove unwanted entries from clients list
      HashMap<String, Cluster.Client> memberClientsHMapNew = new HashMap<>();
      for (Map.Entry<String, Cluster.Client> entry : memberClientsHMap.entrySet()) {
        String clientId = entry.getKey();
        if (memberClientsHM.get(clientId) != null) {
          memberClientsHMapNew.put(clientId, memberClientsHMap.get(clientId));
        }
      }
      // replace existing memberClientsHMap by memberClientsHMapNew
      setMemberClientsHMap(memberClientsHMapNew);

      // update last update time
      Cluster.LAST_UPDATE_TIME = systemNanoTime;
    }
  }

  /**
   * Member Inner Class
   *
   *
   */
  public static class Statement {

    private String queryDefn;
    private long numTimesCompiled;
    private long numExecution;
    private long numExecutionsInProgress;
    private long numTimesGlobalIndexLookup;
    private long numRowsModified;
    private long parseTime;
    private long bindTime;
    private long optimizeTime;
    private long routingInfoTime;
    private long generateTime;
    private long totalCompilationTime;
    private long executionTime;
    private long projectionTime;
    private long totalExecutionTime;
    private long rowsModificationTime;
    private long qNNumRowsSeen;
    private long qNMsgSendTime;
    private long qNMsgSerTime;
    private long qNRespDeSerTime;

    public static String[] getGridColumnNames() {
      return new String[] {PulseConstants.MBEAN_COLNAME_QUERYDEFINITION,
          PulseConstants.MBEAN_COLNAME_NUMEXECUTION,
          PulseConstants.MBEAN_COLNAME_TOTALEXECUTIONTIME,
          PulseConstants.MBEAN_COLNAME_NUMEXECUTIONSINPROGRESS,
          PulseConstants.MBEAN_COLNAME_NUMTIMESCOMPILED,
          PulseConstants.MBEAN_COLNAME_NUMTIMESGLOBALINDEXLOOKUP,
          PulseConstants.MBEAN_COLNAME_NUMROWSMODIFIED, PulseConstants.MBEAN_COLNAME_PARSETIME,
          PulseConstants.MBEAN_COLNAME_BINDTIME, PulseConstants.MBEAN_COLNAME_OPTIMIZETIME,
          PulseConstants.MBEAN_COLNAME_ROUTINGINFOTIME, PulseConstants.MBEAN_COLNAME_GENERATETIME,
          PulseConstants.MBEAN_COLNAME_TOTALCOMPILATIONTIME,
          PulseConstants.MBEAN_COLNAME_EXECUTIONTIME, PulseConstants.MBEAN_COLNAME_PROJECTIONTIME,
          PulseConstants.MBEAN_COLNAME_ROWSMODIFICATIONTIME,
          PulseConstants.MBEAN_COLNAME_QNNUMROWSSEEN, PulseConstants.MBEAN_COLNAME_QNMSGSENDTIME,
          PulseConstants.MBEAN_COLNAME_QNMSGSERTIME, PulseConstants.MBEAN_COLNAME_QNRESPDESERTIME};
    }

    public static String[] getGridColumnAttributes() {
      return new String[] {PulseConstants.MBEAN_ATTRIBUTE_QUERYDEFINITION,
          PulseConstants.MBEAN_ATTRIBUTE_NUMEXECUTION,
          PulseConstants.MBEAN_ATTRIBUTE_TOTALEXECUTIONTIME,
          PulseConstants.MBEAN_ATTRIBUTE_NUMEXECUTIONSINPROGRESS,
          PulseConstants.MBEAN_ATTRIBUTE_NUMTIMESCOMPILED,
          PulseConstants.MBEAN_ATTRIBUTE_NUMTIMESGLOBALINDEXLOOKUP,
          PulseConstants.MBEAN_ATTRIBUTE_NUMROWSMODIFIED, PulseConstants.MBEAN_ATTRIBUTE_PARSETIME,
          PulseConstants.MBEAN_ATTRIBUTE_BINDTIME, PulseConstants.MBEAN_ATTRIBUTE_OPTIMIZETIME,
          PulseConstants.MBEAN_ATTRIBUTE_ROUTINGINFOTIME,
          PulseConstants.MBEAN_ATTRIBUTE_GENERATETIME,
          PulseConstants.MBEAN_ATTRIBUTE_TOTALCOMPILATIONTIME,
          PulseConstants.MBEAN_ATTRIBUTE_EXECUTIONTIME,
          PulseConstants.MBEAN_ATTRIBUTE_PROJECTIONTIME,

          PulseConstants.MBEAN_ATTRIBUTE_ROWSMODIFICATIONTIME,
          PulseConstants.MBEAN_ATTRIBUTE_QNNUMROWSSEEN,
          PulseConstants.MBEAN_ATTRIBUTE_QNMSGSENDTIME, PulseConstants.MBEAN_ATTRIBUTE_QNMSGSERTIME,
          PulseConstants.MBEAN_ATTRIBUTE_QNRESPDESERTIME};
    }

    public static int[] getGridColumnWidths() {
      return new int[] {300, 150, 160, 180, 150, 200, 150, 130, 130, 160, 140, 180, 170,
          160, 130, 190, 170, 170, 170, 200};
    }

    /**
     * @return the numTimesCompiled
     */
    public String getQueryDefinition() {
      return queryDefn;
    }

    /**
     * @param queryDefn the query to set
     */
    public void setQueryDefinition(String queryDefn) {
      this.queryDefn = queryDefn;
    }

    /**
     * @return the numTimesCompiled
     */
    public long getNumTimesCompiled() {
      return numTimesCompiled;
    }

    /**
     * @param numTimesCompiled the numTimesCompiled to set
     */
    public void setNumTimesCompiled(long numTimesCompiled) {
      this.numTimesCompiled = numTimesCompiled;
    }

    /**
     * @return the numExecution
     */
    public long getNumExecution() {
      return numExecution;
    }

    /**
     * @param numExecution the numExecution to set
     */
    public void setNumExecution(long numExecution) {
      this.numExecution = numExecution;
    }

    /**
     * @return the numExecutionsInProgress
     */
    public long getNumExecutionsInProgress() {
      return numExecutionsInProgress;
    }

    /**
     * @param numExecutionsInProgress the numExecutionsInProgress to set
     */
    public void setNumExecutionsInProgress(long numExecutionsInProgress) {
      this.numExecutionsInProgress = numExecutionsInProgress;
    }

    /**
     * @return the numTimesGlobalIndexLookup
     */
    public long getNumTimesGlobalIndexLookup() {
      return numTimesGlobalIndexLookup;
    }

    /**
     * @param numTimesGlobalIndexLookup the numTimesGlobalIndexLookup to set
     */
    public void setNumTimesGlobalIndexLookup(long numTimesGlobalIndexLookup) {
      this.numTimesGlobalIndexLookup = numTimesGlobalIndexLookup;
    }

    /**
     * @return the numRowsModified
     */
    public long getNumRowsModified() {
      return numRowsModified;
    }

    /**
     * @param numRowsModified the numRowsModified to set
     */
    public void setNumRowsModified(long numRowsModified) {
      this.numRowsModified = numRowsModified;
    }

    /**
     * @return the parseTime
     */
    public long getParseTime() {
      return parseTime;
    }

    /**
     * @param parseTime the parseTime to set
     */
    public void setParseTime(long parseTime) {
      this.parseTime = parseTime;
    }

    /**
     * @return the bindTime
     */
    public long getBindTime() {
      return bindTime;
    }

    /**
     * @param bindTime the bindTime to set
     */
    public void setBindTime(long bindTime) {
      this.bindTime = bindTime;
    }

    /**
     * @return the optimizeTime
     */
    public long getOptimizeTime() {
      return optimizeTime;
    }

    /**
     * @param optimizeTime the optimizeTime to set
     */
    public void setOptimizeTime(long optimizeTime) {
      this.optimizeTime = optimizeTime;
    }

    /**
     * @return the routingInfoTime
     */
    public long getRoutingInfoTime() {
      return routingInfoTime;
    }

    /**
     * @param routingInfoTime the routingInfoTime to set
     */
    public void setRoutingInfoTime(long routingInfoTime) {
      this.routingInfoTime = routingInfoTime;
    }

    /**
     * @return the generateTime
     */
    public long getGenerateTime() {
      return generateTime;
    }

    /**
     * @param generateTime the generateTime to set
     */
    public void setGenerateTime(long generateTime) {
      this.generateTime = generateTime;
    }

    /**
     * @return the totalCompilationTime
     */
    public long getTotalCompilationTime() {
      return totalCompilationTime;
    }

    /**
     * @param totalCompilationTime the totalCompilationTime to set
     */
    public void setTotalCompilationTime(long totalCompilationTime) {
      this.totalCompilationTime = totalCompilationTime;
    }

    /**
     * @return the executionTime
     */
    public long getExecutionTime() {
      return executionTime;
    }

    /**
     * @param executionTime the executionTime to set
     */
    public void setExecutionTime(long executionTime) {
      this.executionTime = executionTime;
    }

    /**
     * @return the projectionTime
     */
    public long getProjectionTime() {
      return projectionTime;
    }

    /**
     * @param projectionTime the projectionTime to set
     */
    public void setProjectionTime(long projectionTime) {
      this.projectionTime = projectionTime;
    }

    /**
     * @return the totalExecutionTime
     */
    public long getTotalExecutionTime() {
      return totalExecutionTime;
    }

    /**
     * @param totalExecutionTime the totalExecutionTime to set
     */
    public void setTotalExecutionTime(long totalExecutionTime) {
      this.totalExecutionTime = totalExecutionTime;
    }

    /**
     * @return the rowsModificationTime
     */
