public GrpcChannel acquireChannel(GrpcNetworkGroup networkGroup,
      GrpcServerAddress serverAddress, AlluxioConfiguration conf, boolean alwaysEnableTLS) {
    GrpcChannelKey channelKey = getChannelKey(networkGroup, serverAddress, conf);
    CountingReference<ManagedChannel> channelRef =
        mChannels.compute(channelKey, (key, ref) -> {
          boolean shutdownExistingConnection = false;
          int existingRefCount = 0;
          if (ref != null) {
            // Connection exists, wait for health check.
            if (waitForConnectionReady(ref.get(), conf)) {
              LOG.debug("Acquiring an existing connection. ConnectionKey: {}. Ref-count: {}", key,
                  ref.getRefCount());

              return ref.reference();
            } else {
              // Health check failed.
              shutdownExistingConnection = true;
            }
          }
          // Existing connection should be shutdown.
          if (shutdownExistingConnection) {
            existingRefCount = ref.getRefCount();
            LOG.debug("Shutting down an existing unhealthy connection. "
                + "ConnectionKey: {}. Ref-count: {}", key, existingRefCount);
            // Shutdown the channel forcefully as it's already unhealthy.
            shutdownManagedChannel(ref.get());
          }

          // Create a new managed channel.
          LOG.debug("Creating a new managed channel. ConnectionKey: {}. Ref-count:{},"
                  + " alwaysEnableTLS:{} config TLS:{}", key, existingRefCount, alwaysEnableTLS,
              conf.getBoolean(alluxio.conf.PropertyKey.NETWORK_TLS_ENABLED));
          ManagedChannel managedChannel = createManagedChannel(channelKey, conf, alwaysEnableTLS);
          // Set map reference.
          return new CountingReference<>(managedChannel, existingRefCount).reference();
        });

    return new GrpcChannel(channelKey, channelRef.get());
  }
@Test
  public void testRoundRobin() throws Exception {
    int streamingGroupSize =
        sConf.getInt(PropertyKey.USER_NETWORK_STREAMING_MAX_CONNECTIONS);

    try (CloseableTestServer server = createServer()) {
      List<GrpcServerAddress> addresses = new ArrayList<>(streamingGroupSize);
      // Create channel keys.
      for (int i = 0; i < streamingGroupSize; i++) {
        addresses.add(server.getConnectAddress());
      }
      // Acquire connections.
      List<GrpcChannel> connections =
          addresses.stream()
              .map(address -> GrpcChannelPool.INSTANCE.acquireChannel(
                  GrpcNetworkGroup.STREAMING, address, sConf, false))
              .collect(Collectors.toList());

      // Validate all are different.
      Assert.assertEquals(streamingGroupSize, connections.stream().distinct().count());
    }
  }