protected InetSocketAddress getBroadcastRpcAddress()

in core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultTopologyMonitor.java [500:577]

• 54 lines of code
• 14 McCabe index (conditional complexity)

  protected InetSocketAddress getBroadcastRpcAddress(
      @NonNull AdminRow row, @NonNull EndPoint localEndPoint) {

    InetAddress broadcastRpcInetAddress = null;
    Iterator<String> addrCandidates =
        Iterators.forArray(
            // in system.peers_v2 (Cassandra >= 4.0)
            "native_address",
            // DSE 6.8 introduced native_transport_address and native_transport_port for the
            // listen address.
            "native_transport_address",
            // in system.peers or system.local
            "rpc_address");

    while (broadcastRpcInetAddress == null && addrCandidates.hasNext())
      broadcastRpcInetAddress = row.getInetAddress(addrCandidates.next());
    // This could only happen if system tables are corrupted, but handle gracefully
    if (broadcastRpcInetAddress == null) {
      LOG.warn(
          "[{}] Unable to determine broadcast RPC IP address, returning null.  "
              + "This is likely due to a misconfiguration or invalid system tables.  "
              + "Please validate the contents of system.local and/or {}.",
          logPrefix,
          getPeerTableName());
      return null;
    }

    Integer broadcastRpcPort = null;
    Iterator<String> portCandidates =
        Iterators.forArray(
            // in system.peers_v2 (Cassandra >= 4.0)
            NATIVE_PORT,
            // DSE 6.8 introduced native_transport_address and native_transport_port for the
            // listen address.
            NATIVE_TRANSPORT_PORT,
            // system.local for Cassandra >= 4.0
            "rpc_port");

    while ((broadcastRpcPort == null || broadcastRpcPort == 0) && portCandidates.hasNext()) {

      String colName = portCandidates.next();
      broadcastRpcPort = row.getInteger(colName);
      // Support override for SSL port (if enabled) in DSE
      if (NATIVE_TRANSPORT_PORT.equals(colName) && context.getSslEngineFactory().isPresent()) {

        String sslColName = colName + "_ssl";
        broadcastRpcPort = row.getInteger(sslColName);
      }
    }
    // use the default port if no port information was found in the row;
    // note that in rare situations, the default port might not be known, in which case we
    // report zero, as advertised in the javadocs of Node and NodeInfo.
    if (broadcastRpcPort == null || broadcastRpcPort == 0) {

      LOG.warn(
          "[{}] Unable to determine broadcast RPC port.  "
              + "Trying to fall back to port used by the control connection.",
          logPrefix);
      broadcastRpcPort = port == -1 ? 0 : port;
    }

    InetSocketAddress broadcastRpcAddress =
        new InetSocketAddress(broadcastRpcInetAddress, broadcastRpcPort);
    if (row.contains("peer") && broadcastRpcAddress.equals(localEndPoint.resolve())) {
      // JAVA-2303: if the peer is actually the control node, ignore that peer as it is likely
      // a misconfiguration problem.
      LOG.warn(
          "[{}] Control node {} has an entry for itself in {}: this entry will be ignored. "
              + "This is likely due to a misconfiguration; please verify your rpc_address "
              + "configuration in cassandra.yaml on all nodes in your cluster.",
          logPrefix,
          localEndPoint,
          getPeerTableName());
      return null;
    }

    return broadcastRpcAddress;
  }