szetszwo commented on code in PR #1534:
URL: https://github.com/apache/ratis/pull/1534#discussion_r3707987084
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientReplies.java:
##########
@@ -60,6 +60,19 @@ class ReplyMap {
ReplyEntry submitRequest(RequestEntry requestEntry, boolean isClose,
CompletableFuture<DataStreamReply> f) {
LOG.debug("put {} to the map for {}", requestEntry, clientInvocationId);
+ if (requestEntry.type == Type.STREAM_COMMAND) {
+ if (map.containsKey(requestEntry)) {
+ final IllegalStateException exception = new IllegalStateException(
+ "A STREAM_COMMAND is already pending for " + requestEntry
+ + " for " + clientInvocationId
+ + "; wait for the previous command reply before sending
another");
+ f.completeExceptionally(exception);
+ return null;
+ }
+ final ReplyEntry entry = new ReplyEntry(isClose, f);
+ map.put(requestEntry, entry);
+ return entry;
+ }
// ConcurrentHashMap.computeIfAbsent javadoc: the function is applied at
most once per key.
return map.computeIfAbsent(requestEntry, r -> new ReplyEntry(isClose,
f));
Review Comment:
```java
ReplyEntry submitRequest(RequestEntry requestEntry, boolean isClose,
CompletableFuture<DataStreamReply> f) {
LOG.debug("put {} to the map for {}", requestEntry,
clientInvocationId);
final MemoizedSupplier<ReplyEntry> supplier =
MemoizedSupplier.valueOf(() -> new ReplyEntry(isClose, f));
final ReplyEntry reply = map.computeIfAbsent(requestEntry, r ->
supplier.get());
if (requestEntry.type == Type.STREAM_COMMAND &&
!supplier.isInitialized()) {
final IllegalStateException exception = new IllegalStateException(
"STREAM_COMMAND already exist: " + requestEntry + " for " +
clientInvocationId);
f.completeExceptionally(exception);
return null;
}
return reply;
}
```
- We should only make one call to the map as above; otherwise, it is not
atomic. In this case, we may use MemoizedSupplier.
- Let's also make the exception message shorter; see below
<img width="562" height="369" alt="Image"
src="https://github.com/user-attachments/assets/8d53c239-1492-47ba-a929-e1fa2a1cadb1";
/>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
