RATIS-2325. Create GrpcStubPool for GrpcServerProtocolClient

[symious]

What changes were proposed in this pull request?

This is to fix the comment mentioned in #1283

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2325

How was this patch tested?

Ozone Follower Read performance.

[symious]

@szetszwo Updated the client's configuration, could you help to check if the warning still appears?

[szetszwo]

... Updated the client's configuration, ...

@symious , why not using the same configuration that we are current using?

[szetszwo]

@symious , thanks for the update! Please see the comments inlined and also https://issues.apache.org/jira/secure/attachment/13079229/1306_review.patch

[symious]

why not using the same configuration that we are current using?

@szetszwo In #1283, you mentioned the warning logs. The configuration change is to resolve them.

[szetszwo]

Yiyang, @symious , thanks for the update!

Just a minor comment inlined.

[szetszwo]

We should check it is already initialized. If not, it should not be initialized and then shut down.

[szetszwo]

... could you help to check if the warning still appears?

Checked that the gRPC warnings have disappeared after the change. Thanks.

[szetszwo]

@symious , thanks for the update. Found one more problem. Please see the comment inlined.

[szetszwo]

We should memoize everything here. Otherwise, it will build the channel but not the stub. See below and also https://issues.apache.org/jira/secure/attachment/13079229/1306_review.patch

  static ManagedChannel buildManagedChannel(String address, SslContext sslContext) {
    NettyChannelBuilder channelBuilder = NettyChannelBuilder.forTarget(address)
        .keepAliveTime(10, TimeUnit.MINUTES)
        .keepAliveWithoutCalls(false)
        .idleTimeout(30, TimeUnit.MINUTES)
        .withOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(64 << 10, 128 << 10));
    if (sslContext != null) {
      LOG.debug("Setting TLS for {}", address);
      channelBuilder.useTransportSecurity().sslContext(sslContext);
    } else {
      channelBuilder.negotiationType(NegotiationType.PLAINTEXT);
    }
    ManagedChannel ch = channelBuilder.build();
    ch.getState(true);
    return ch;
  }

  static final class Stub<S extends AbstractStub<S>> {
    private final ManagedChannel ch;
    private final S stub;
    private final Semaphore permits;

    Stub(String address, SslContext sslContext, Function<ManagedChannel, S> stubFactory, int maxInflight) {
      this.ch = buildManagedChannel(address, sslContext);
      this.stub = stubFactory.apply(ch);
      this.permits = new Semaphore(maxInflight);
    }

    ...
  }

  GrpcStubPool(int connections, String address, SslContext sslContext, Function<ManagedChannel, S> stubFactory,
      int maxInflightPerConn) {
    Preconditions.assertTrue(connections > 1, "connections must be > 1");
    final List<MemoizedSupplier<Stub<S>>> pool = new ArrayList<>(connections);
    for (int i = 0; i < connections; i++) {
      pool.add(MemoizedSupplier.valueOf(() -> new Stub<>(address, sslContext, stubFactory, maxInflightPerConn)));
    }
    this.pool = Collections.unmodifiableList(pool);
  }

[szetszwo]

+1 the change looks good.