Copilot commented on code in PR #10767:
URL: https://github.com/apache/rocketmq/pull/10767#discussion_r3701685632
##########
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java:
##########
@@ -235,52 +236,71 @@ protected CompletableFuture<Void>
processGetConsumerRunningInfo(RemotingCommand
if (Objects.isNull(header) || !header.isJstackEnable()) {
return CompletableFuture.completedFuture(null);
}
- this.writeTelemetryCommand(TelemetryCommand.newBuilder()
+ String nonce =
this.grpcChannelManager.addResponseFuture(responseFuture);
+ boolean written =
this.writeTelemetryCommand(TelemetryCommand.newBuilder()
.setPrintThreadStackTraceCommand(PrintThreadStackTraceCommand.newBuilder()
-
.setNonce(this.grpcChannelManager.addResponseFuture(responseFuture))
+ .setNonce(nonce)
.build())
.build());
+ if (!written) {
+ this.completeResponseFutureOnWriteFailure(nonce, responseFuture);
+ }
return CompletableFuture.completedFuture(null);
}
@Override
protected CompletableFuture<Void>
processConsumeMessageDirectly(RemotingCommand command,
ConsumeMessageDirectlyResultRequestHeader header,
MessageExt messageExt,
CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>>
responseFuture) {
- this.writeTelemetryCommand(TelemetryCommand.newBuilder()
+ String nonce =
this.grpcChannelManager.addResponseFuture(responseFuture);
+ boolean written =
this.writeTelemetryCommand(TelemetryCommand.newBuilder()
.setVerifyMessageCommand(VerifyMessageCommand.newBuilder()
-
.setNonce(this.grpcChannelManager.addResponseFuture(responseFuture))
+ .setNonce(nonce)
.setMessage(GrpcConverter.getInstance().buildMessage(messageExt))
.build())
.build());
+ if (!written) {
+ this.completeResponseFutureOnWriteFailure(nonce, responseFuture);
+ }
return CompletableFuture.completedFuture(null);
}
public String getClientId() {
return clientId;
}
- public void writeTelemetryCommand(TelemetryCommand command) {
+ public boolean writeTelemetryCommand(TelemetryCommand command) {
StreamObserver<TelemetryCommand> observer =
this.telemetryCommandRef.get();
Review Comment:
`writeTelemetryCommand` is a public method and changing its signature from
`void` to `boolean` is a binary-incompatible API change (downstream code
compiled against the old signature will break at runtime). To preserve
compatibility, consider keeping `public void
writeTelemetryCommand(TelemetryCommand)` and introducing a new internal method
(e.g. `tryWriteTelemetryCommand`) that returns `boolean` for the fail-fast
logic in this class.
##########
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannelTest.java:
##########
@@ -79,4 +96,58 @@ public void testChannelExtendAttributeParse() {
assertEquals(clientSettings,
GrpcClientChannel.parseChannelExtendAttribute(this.grpcClientChannel));
assertNull(GrpcClientChannel.parseChannelExtendAttribute(mock(RemotingChannel.class)));
}
-}
\ No newline at end of file
+
+ @Test
+ public void
testGetConsumerRunningInfoShouldFailFastWhenObserverIsMissing() throws
Exception {
+ CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>>
responseFuture = new CompletableFuture<>();
+
when(grpcChannelManager.addResponseFuture(eq(responseFuture))).thenReturn("nonce-1");
+
when(grpcChannelManager.getAndRemoveResponseFuture(eq("nonce-1"))).thenReturn((CompletableFuture)
responseFuture);
Review Comment:
Avoid the unchecked raw cast when stubbing `getAndRemoveResponseFuture`; it
hides type issues and produces compiler warnings. Mockito supports specifying
the generic type parameter explicitly here.
This issue also appears on line 124 of the same file.
--
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]
