@Override
    public void connect(Configuration conf) throws IOException {
      String mcAddress =
        conf.get(HConstants.STATUS_MULTICAST_ADDRESS, HConstants.DEFAULT_STATUS_MULTICAST_ADDRESS);
      int port =
        conf.getInt(HConstants.STATUS_MULTICAST_PORT, HConstants.DEFAULT_STATUS_MULTICAST_PORT);
      String bindAddress = conf.get(HConstants.STATUS_MULTICAST_PUBLISHER_BIND_ADDRESS,
        HConstants.DEFAULT_STATUS_MULTICAST_PUBLISHER_BIND_ADDRESS);
      String niName = conf.get(HConstants.STATUS_MULTICAST_NI_NAME);

      final InetAddress ina;
      try {
        ina = InetAddress.getByName(mcAddress);
      } catch (UnknownHostException e) {
        close();
        throw new IOException("Can't connect to " + mcAddress, e);
      }
      final InetSocketAddress isa = new InetSocketAddress(mcAddress, port);

      InternetProtocolFamily family;
      NetworkInterface ni;
      if (niName != null) {
        if (ina instanceof Inet6Address) {
          family = InternetProtocolFamily.IPv6;
        } else {
          family = InternetProtocolFamily.IPv4;
        }
        ni = NetworkInterface.getByName(niName);
      } else {
        InetAddress localAddress;
        if (ina instanceof Inet6Address) {
          localAddress = Addressing.getIp6Address();
          family = InternetProtocolFamily.IPv6;
        } else {
          localAddress = Addressing.getIp4Address();
          family = InternetProtocolFamily.IPv4;
        }
        ni = NetworkInterface.getByInetAddress(localAddress);
      }
      Bootstrap b = new Bootstrap();
      b.group(group)
        .channelFactory(new HBaseDatagramChannelFactory<Channel>(NioDatagramChannel.class, family))
        .option(ChannelOption.SO_REUSEADDR, true).handler(new ClusterMetricsEncoder(isa));
      try {
        LOG.debug("Channel bindAddress={}, networkInterface={}, INA={}", bindAddress, ni, ina);
        channel = (DatagramChannel) b.bind(bindAddress, 0).sync().channel();
        channel.joinGroup(ina, ni, null, channel.newPromise()).sync();
        channel.connect(isa).sync();
        // Set into configuration in case many networks available. Do this for tests so that
        // server and client use same Interface (presuming share same Configuration).
        // TestAsyncTableRSCrashPublish was failing when connected to VPN because extra networks
        // available with Master binding on one Interface and client on another so test failed.
        if (ni != null) {
          conf.set(HConstants.STATUS_MULTICAST_NI_NAME, ni.getName());
        }
      } catch (InterruptedException e) {
        close();
        throw ExceptionUtil.asInterrupt(e);
      }
    }
Get the servers which died since a given timestamp. protected because it can be subclassed by
the tests.
hbase-server/src/main/java/org/apache/hadoop/hbase/master/ClusterStatusPublisher.java