[Bug][HStore Client] gRPC stub pools bind every entry to one ManagedChannel

[bitflicker64]

HugeGraph version

Current master, verified at commit b9710a7b0319ace41dd3c4ff5fb1165f53ee538e.

Problem

AbstractGrpcClient creates an array of 32 ManagedChannel instances per target, apparently to distribute requests across a channel pool. However, both the blocking-stub and asynchronous-stub initialization loops bind every array entry to the same channel.

The loop variable is i, but each iteration selects channels[index], where index was calculated once before the loop:

int index = (int) (l & (concurrency - 1));

HgPair<ManagedChannel, AbstractBlockingStub>[] value = new HgPair[concurrency];
IntStream.range(0, concurrency).forEach(i -> {
    ManagedChannel channel = channels[index];
    AbstractBlockingStub stub = getBlockingStub(channel);
    value[i] = new HgPair<>(channel, stub);
});

The asynchronous path contains the same indexing error:

IntStream.range(0, concurrency).parallel().forEach(i -> {
    ManagedChannel channel = channels[index];
    AbstractAsyncStub stub = getAsyncStub(channel);
    value[i] = new HgPair<>(channel, stub);
});

Source:

hugegraph/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java

Lines 93 to 161 in b9710a7

	public AbstractBlockingStub getBlockingStub(String target) {
	ManagedChannel[] channels = getChannels(target);
	HgPair<ManagedChannel, AbstractBlockingStub>[] pairs = blockingStubs.get(target);
	long l = counter.getAndIncrement();
	if (l >= limit) {
	counter.set(0);
	}
	int index = (int) (l & (concurrency - 1));
	if (pairs == null) {
	synchronized (blockingStubs) {
	pairs = blockingStubs.get(target);
	if (pairs == null) {
	HgPair<ManagedChannel, AbstractBlockingStub>[] value = new HgPair[concurrency];
	IntStream.range(0, concurrency).forEach(i -> {
	ManagedChannel channel = channels[index];
	AbstractBlockingStub stub = getBlockingStub(channel);
	value[i] = new HgPair<>(channel, stub);
	// log.info("create channel for {}",target);
	});
	blockingStubs.put(target, value);
	AbstractBlockingStub stub = value[index].getValue();
	return (AbstractBlockingStub) setBlockingStubOption(stub);
	}
	}
	}
	return (AbstractBlockingStub) setBlockingStubOption(pairs[index].getValue());
	}
	private AbstractStub setBlockingStubOption(AbstractBlockingStub stub) {
	return stub.withDeadlineAfter(config.getGrpcTimeoutSeconds(), TimeUnit.SECONDS)
	.withMaxInboundMessageSize(
	config.getGrpcMaxInboundMessageSize())
	.withMaxOutboundMessageSize(
	config.getGrpcMaxOutboundMessageSize());
	}
	public AbstractAsyncStub getAsyncStub(ManagedChannel channel) {
	return null;
	}
	public AbstractAsyncStub getAsyncStub(String target) {
	ManagedChannel[] channels = getChannels(target);
	HgPair<ManagedChannel, AbstractAsyncStub>[] pairs = asyncStubs.get(target);
	long l = counter.getAndIncrement();
	if (l >= limit) {
	counter.set(0);
	}
	int index = (int) (l & (concurrency - 1));
	if (pairs == null) {
	synchronized (asyncStubs) {
	pairs = asyncStubs.get(target);
	if (pairs == null) {
	HgPair<ManagedChannel, AbstractAsyncStub>[] value = new HgPair[concurrency];
	IntStream.range(0, concurrency).parallel().forEach(i -> {
	ManagedChannel channel = channels[index];
	AbstractAsyncStub stub = getAsyncStub(channel);
	// stub.withMaxInboundMessageSize(config.getGrpcMaxInboundMessageSize())
	// .withMaxOutboundMessageSize(config.getGrpcMaxOutboundMessageSize());
	value[i] = new HgPair<>(channel, stub);
	// log.info("create channel for {}",target);
	});
	asyncStubs.put(target, value);
	AbstractAsyncStub stub =
	(AbstractAsyncStub) setStubOption(value[index].getValue());
	return stub;
	}
	}
	}
	return (AbstractAsyncStub) setStubOption(pairs[index].getValue());

Actual behavior

Although getChannels(target) creates 32 channels, all entries in a newly initialized blocking or asynchronous stub array reference whichever single channel happened to be selected by the first call's index.

Subsequent calls rotate across the stub-array indexes, but those stubs all use the same underlying ManagedChannel. The remaining 31 created channels are unused by that stub pool.

Consequences include:

• The intended channel-level request distribution does not occur.
• All traffic for that client and target is concentrated on one channel.
• Creating 31 additional channels consumes resources without providing concurrency or isolation.
• A problem affecting the selected channel affects every entry in the corresponding stub pool.

This defect is independent of DNS caching or Kubernetes Store replacement behavior and can be verified without a cluster.

Expected behavior

Each stub-pool entry should be paired with its corresponding channel-pool entry:

ManagedChannel channel = channels[i];

After initialization, the 32 stub entries should collectively reference all 32 channels created for the target.

Minimal fix

Change both initialization paths from:

ManagedChannel channel = channels[index];

to:

ManagedChannel channel = channels[i];

Suggested test

Add a unit test using a test subclass of AbstractGrpcClient that records the ManagedChannel passed to getBlockingStub(ManagedChannel) and getAsyncStub(ManagedChannel) during pool initialization.

For a new target, assert that:

• each stub factory is called concurrency times;
• the recorded channel identities contain concurrency distinct entries; and
• repeated public stub lookups rotate across stubs backed by those distinct channels.

[dosubot]

Good catch — I can confirm the bug. In both getBlockingStub and getAsyncStub initialization paths, channels[index] is used where channels[i] should be, so all concurrency stub entries end up bound to the same ManagedChannel [1] [2].

I didn't find any existing PR that fixes this. The related issue #3124 you filed does mention this as a "separate channel-pool indexing defect" to be tracked independently, so this issue is a good standalone report for it [3].

The fix (channels[index] → channels[i]) and the suggested test approach both look correct. One additional thing worth noting: the async path uses .parallel() on the IntStream [4], so a test for that variant should also verify thread-safety of the recorded channel identities (e.g., use a ConcurrentHashMap or similar to collect them).

To reply, just mention @dosu.

---

Docs are dead. Just use Dosu.