This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 9fe1449d [ISSUE #1309] [Java] Improve gRPC transport resilience (#1310)
9fe1449d is described below
commit 9fe1449d19449b41442aa3a97ab168ed6b5bd6b1
Author: qianye <[email protected]>
AuthorDate: Wed Jul 29 14:10:37 2026 +0800
[ISSUE #1309] [Java] Improve gRPC transport resilience (#1310)
---
.../apache/rocketmq/client/java/impl/Client.java | 9 +-
.../rocketmq/client/java/impl/ClientImpl.java | 24 +-
.../rocketmq/client/java/impl/ClientManager.java | 7 +
.../client/java/impl/ClientManagerImpl.java | 126 +++++++++
.../client/java/impl/ClientSessionImpl.java | 91 +++++--
.../apache/rocketmq/client/java/rpc/RpcClient.java | 14 +
.../rocketmq/client/java/rpc/RpcClientImpl.java | 11 +
.../apache/rocketmq/client/java/rpc/RpcFuture.java | 26 +-
.../rocketmq/client/java/impl/ClientImplTest.java | 16 +-
.../client/java/impl/ClientManagerImplTest.java | 167 +++++++++++-
.../client/java/impl/ClientSessionImplTest.java | 141 +++++++++-
.../rocketmq/client/java/rpc/RpcFutureTest.java | 79 ++++++
...ProducerHalfOpenTcpRecoveryIntegrationTest.java | 296 +++++++++++++++++++++
13 files changed, 975 insertions(+), 32 deletions(-)
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
index 4de14ffa..83510113 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
@@ -55,6 +55,13 @@ public interface Client {
*/
boolean isSslEnabled();
+ /**
+ * Reconnect telemetry to the specified endpoints.
+ *
+ * @param endpoints endpoints to reconnect.
+ */
+ void reconnectTelemetry(Endpoints endpoints);
+
/**
* Send Heartbeat
*
@@ -75,4 +82,4 @@ public interface Client {
* <p>Perform some statistics for the client.
*/
void doStats();
-}
\ No newline at end of file
+}
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
index 4d772104..b523af73 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
@@ -130,7 +130,6 @@ public abstract class ClientImpl extends
AbstractIdleService implements Client,
private final ReadWriteLock sessionsLock;
private final CompositedMessageInterceptor compositedMessageInterceptor;
- private boolean receiveReconnect = false;
public ClientImpl(ClientConfiguration clientConfiguration, Set<String>
topics) {
this.clientConfiguration = checkNotNull(clientConfiguration,
"clientConfiguration should not be null");
@@ -294,7 +293,7 @@ public abstract class ClientImpl extends
AbstractIdleService implements Client,
@Override
public void onReconnectEndpointsCommand(Endpoints endpoints,
ReconnectEndpointsCommand command) {
- receiveReconnect = true;
+ getClientManager().reconnect(endpoints);
}
/**
@@ -547,12 +546,21 @@ public abstract class ClientImpl extends
AbstractIdleService implements Client,
return clientId;
}
- public boolean isReceiveReconnect() {
- return receiveReconnect;
- }
-
- public void setReceiveReconnect(boolean receiveReconnect) {
- this.receiveReconnect = receiveReconnect;
+ @Override
+ public void reconnectTelemetry(Endpoints endpoints) {
+ final ClientSessionImpl clientSession;
+ sessionsLock.readLock().lock();
+ try {
+ clientSession = sessionsTable.get(endpoints);
+ } finally {
+ sessionsLock.readLock().unlock();
+ }
+ if (null == clientSession) {
+ log.warn("Failed to rebuild telemetry because client session does
not exist, endpoints={}, clientId={}",
+ endpoints, clientId);
+ return;
+ }
+ clientSession.reconnect();
}
/**
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManager.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManager.java
index d8f5c22f..ea19ea89 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManager.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManager.java
@@ -62,6 +62,13 @@ public abstract class ClientManager extends
AbstractIdleService {
*/
public abstract ScheduledExecutorService getScheduler();
+ /**
+ * Reconnect to the specified endpoints.
+ *
+ * @param endpoints endpoints to reconnect.
+ */
+ public abstract void reconnect(Endpoints endpoints);
+
/**
* Query topic route asynchronously, the method ensures no throwable.
*
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
index d0fd49bb..c7b3e1bb 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
@@ -42,9 +42,14 @@ import apache.rocketmq.v2.SendMessageResponse;
import apache.rocketmq.v2.SyncLiteSubscriptionRequest;
import apache.rocketmq.v2.SyncLiteSubscriptionResponse;
import apache.rocketmq.v2.TelemetryCommand;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import com.google.errorprone.annotations.concurrent.GuardedBy;
+import io.grpc.ConnectivityState;
import io.grpc.Metadata;
+import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.time.Duration;
@@ -52,12 +57,16 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.net.ssl.SSLException;
@@ -94,6 +103,9 @@ public class ClientManagerImpl extends ClientManager {
public static final Duration SYNC_SETTINGS_DELAY = Duration.ofSeconds(1);
public static final Duration SYNC_SETTINGS_PERIOD = Duration.ofMinutes(5);
+ static final int HEART_BEAT_FAILURE_THRESHOLD = 2;
+ static final Duration TRANSPORT_RECOVERY_COOLDOWN = Duration.ofSeconds(30);
+
private static final Logger log =
LoggerFactory.getLogger(ClientManagerImpl.class);
private final Client client;
@@ -101,6 +113,8 @@ public class ClientManagerImpl extends ClientManager {
@GuardedBy("rpcClientTableLock")
private final Map<Endpoints, RpcClient> rpcClientTable;
private final ReadWriteLock rpcClientTableLock;
+ private final ConcurrentMap<Endpoints, Integer> heartbeatFailureAttempts;
+ private final ConcurrentMap<Endpoints, TransportRecoveryState>
transportRecoveryStates;
/**
* In charge of all scheduled tasks.
@@ -116,6 +130,8 @@ public class ClientManagerImpl extends ClientManager {
this.client = client;
this.rpcClientTable = new HashMap<>();
this.rpcClientTableLock = new ReentrantReadWriteLock();
+ this.heartbeatFailureAttempts = new ConcurrentHashMap<>();
+ this.transportRecoveryStates = new ConcurrentHashMap<>();
final long clientIndex = client.getClientId().getIndex();
this.scheduler = new ScheduledThreadPoolExecutor(
Runtime.getRuntime().availableProcessors(),
@@ -148,6 +164,8 @@ public class ClientManagerImpl extends ClientManager {
final Duration idleDuration = rpcClient.idleDuration();
if (idleDuration.compareTo(RPC_CLIENT_MAX_IDLE_DURATION) > 0) {
it.remove();
+ heartbeatFailureAttempts.remove(endpoints);
+ transportRecoveryStates.remove(endpoints);
rpcClient.shutdown();
log.info("Rpc client has been idle for a long time,
endpoints={}, idleDuration={}, " +
"rpcClientMaxIdleDuration={}, clientId={}",
endpoints, idleDuration,
@@ -196,6 +214,30 @@ public class ClientManagerImpl extends ClientManager {
}
}
+ private RpcClient getRpcClientIfPresent(Endpoints endpoints) {
+ rpcClientTableLock.readLock().lock();
+ try {
+ return rpcClientTable.get(endpoints);
+ } finally {
+ rpcClientTableLock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public void reconnect(Endpoints endpoints) {
+ final RpcClient rpcClient = getRpcClientIfPresent(endpoints);
+ if (null == rpcClient) {
+ log.warn("Failed to reconnect because rpc client does not exist,
endpoints={}, clientId={}",
+ endpoints, client.getClientId());
+ return;
+ }
+ reconnect(endpoints, rpcClient);
+ }
+
+ void reconnect(Endpoints endpoints, RpcClient rpcClient) {
+ recoverTransport(endpoints, rpcClient, "server reconnect command",
false);
+ }
+
@Override
public RpcFuture<QueryRouteRequest, QueryRouteResponse>
queryRoute(Endpoints endpoints, QueryRouteRequest request,
Duration duration) {
@@ -219,12 +261,96 @@ public class ClientManagerImpl extends ClientManager {
final Context context = new Context(endpoints, metadata);
final RpcClient rpcClient = getRpcClient(endpoints);
ListenableFuture<HeartbeatResponse> future =
rpcClient.heartbeat(metadata, request, asyncWorker, duration);
+ monitorHeartbeat(endpoints, rpcClient, future);
return new RpcFuture<>(context, request, future);
} catch (Throwable t) {
return new RpcFuture<>(t);
}
}
+ void monitorHeartbeat(Endpoints endpoints, RpcClient rpcClient,
ListenableFuture<HeartbeatResponse> future) {
+ Futures.addCallback(future, new FutureCallback<HeartbeatResponse>() {
+ @Override
+ public void onSuccess(HeartbeatResponse result) {
+ heartbeatFailureAttempts.remove(endpoints);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ final Status.Code code = Status.fromThrowable(t).getCode();
+ if (Status.Code.UNAVAILABLE == code) {
+ heartbeatFailureAttempts.remove(endpoints);
+ if (ConnectivityState.READY == rpcClient.getState(false)) {
+ recoverTransport(endpoints, rpcClient, "heartbeat
failure, statusCode=" + code, true);
+ }
+ return;
+ }
+ if (Status.Code.DEADLINE_EXCEEDED != code) {
+ heartbeatFailureAttempts.remove(endpoints);
+ return;
+ }
+ final int attempts = heartbeatFailureAttempts.merge(endpoints,
1, Integer::sum);
+ if (attempts >= HEART_BEAT_FAILURE_THRESHOLD) {
+ recoverTransport(endpoints, rpcClient, "heartbeat failure,
statusCode=" + code, true);
+ }
+ }
+ }, MoreExecutors.directExecutor());
+ }
+
+ private void recoverTransport(Endpoints endpoints, RpcClient rpcClient,
String reason,
+ boolean respectHeartbeatCooldown) {
+ final TransportRecoveryState recoveryState =
transportRecoveryStates.computeIfAbsent(
+ endpoints, ignored -> new TransportRecoveryState());
+ if (!recoveryState.recovering.compareAndSet(false, true)) {
+ log.info("Skip concurrent transport recovery, endpoints={},
reason={}, clientId={}",
+ endpoints, reason, client.getClientId());
+ return;
+ }
+ try {
+ final long now = System.nanoTime();
+ final long previous = recoveryState.lastRecoveryNanoTime.get();
+ if (TransportRecoveryState.NO_RECOVERY_NANO_TIME != previous) {
+ if (respectHeartbeatCooldown
+ && now - previous < TRANSPORT_RECOVERY_COOLDOWN.toNanos())
{
+ log.info("Skip transport recovery during heartbeat
cooldown, endpoints={}, reason={}, "
+ + "cooldown={}, clientId={}", endpoints, reason,
TRANSPORT_RECOVERY_COOLDOWN,
+ client.getClientId());
+ return;
+ }
+ }
+ recoveryState.lastRecoveryNanoTime.set(now);
+ heartbeatFailureAttempts.remove(endpoints);
+ log.warn("Try to recover transport, endpoints={}, reason={},
clientId={}",
+ endpoints, reason, client.getClientId());
+ try {
+ rpcClient.enterIdle();
+ } catch (RuntimeException e) {
+ log.warn("Failed to enter idle mode while recovering
transport, endpoints={}, reason={}, clientId={}",
+ endpoints, reason, client.getClientId(), e);
+ }
+ try {
+ client.reconnectTelemetry(endpoints);
+ } catch (RuntimeException e) {
+ log.warn("Failed to rebuild telemetry while recovering
transport, endpoints={}, reason={}, "
+ + "clientId={}", endpoints, reason, client.getClientId(),
e);
+ }
+ } finally {
+ recoveryState.recovering.set(false);
+ }
+ }
+
+ private static final class TransportRecoveryState {
+ private static final long NO_RECOVERY_NANO_TIME = Long.MIN_VALUE;
+
+ private final AtomicLong lastRecoveryNanoTime;
+ private final AtomicBoolean recovering;
+
+ private TransportRecoveryState() {
+ this.lastRecoveryNanoTime = new AtomicLong(NO_RECOVERY_NANO_TIME);
+ this.recovering = new AtomicBoolean();
+ }
+ }
+
@Override
public RpcFuture<SendMessageRequest, SendMessageResponse>
sendMessage(Endpoints endpoints,
SendMessageRequest request, Duration duration) {
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientSessionImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientSessionImpl.java
index f0c7de3e..de6fbf36 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientSessionImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientSessionImpl.java
@@ -27,9 +27,11 @@ import apache.rocketmq.v2.VerifyMessageCommand;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
+import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.java.impl.producer.ClientSessionHandler;
import org.apache.rocketmq.client.java.misc.ClientId;
@@ -48,6 +50,8 @@ public class ClientSessionImpl implements
StreamObserver<TelemetryCommand> {
private final ClientSessionHandler sessionHandler;
private final Endpoints endpoints;
private final SettableFuture<Settings> settingsInitFuture;
+ private final Object requestObserverLock;
+ private final AtomicBoolean reconnecting;
private volatile StreamObserver<TelemetryCommand> requestObserver;
@SuppressWarnings("UnstableApiUsage")
@@ -56,6 +60,8 @@ public class ClientSessionImpl implements
StreamObserver<TelemetryCommand> {
this.sessionHandler = sessionHandler;
this.endpoints = endpoints;
this.settingsInitFuture = SettableFuture.create();
+ this.requestObserverLock = new Object();
+ this.reconnecting = new AtomicBoolean();
Futures.withTimeout(settingsInitFuture,
SETTINGS_INITIALIZATION_TIMEOUT.plus(tolerance).toMillis(),
TimeUnit.MILLISECONDS, sessionHandler.getScheduler());
this.requestObserver = sessionHandler.telemetry(endpoints, this);
@@ -68,10 +74,15 @@ public class ClientSessionImpl implements
StreamObserver<TelemetryCommand> {
log.info("Endpoints is deprecated, no longer to renew
requestObserver, endpoints={}, clientId={}",
endpoints, clientId);
sessionHandler.removeClientSession(endpoints, this);
+ reconnecting.set(false);
return;
}
log.info("Try to renew requestObserver, endpoints={},
clientId={}", endpoints, clientId);
- this.requestObserver = sessionHandler.telemetry(endpoints, this);
+ final StreamObserver<TelemetryCommand> requestObserver =
sessionHandler.telemetry(endpoints, this);
+ synchronized (requestObserverLock) {
+ this.requestObserver = requestObserver;
+ reconnecting.set(false);
+ }
} catch (Throwable t) {
log.error("Failed to renew requestObserver, attempt to renew
later, endpoints={}, delay={}, clientId={}",
endpoints, REQUEST_OBSERVER_RENEW_BACKOFF_DELAY, clientId, t);
@@ -99,26 +110,54 @@ public class ClientSessionImpl implements
StreamObserver<TelemetryCommand> {
*/
public void release() {
final ClientId clientId = sessionHandler.getClientId();
- if (null == requestObserver) {
- log.error("[Bug] request observer does not exist, no need to
release, endpoints={}, clientId={}",
- endpoints, clientId);
- return;
- }
- log.info("Begin to release client session, endpoints={}, clientId={}",
endpoints, clientId);
- try {
- requestObserver.onCompleted();
- } catch (Throwable ignore) {
- // Ignore exception on purpose.
+ synchronized (requestObserverLock) {
+ if (null == requestObserver) {
+ log.error("[Bug] request observer does not exist, no need to
release, endpoints={}, clientId={}",
+ endpoints, clientId);
+ return;
+ }
+ log.info("Begin to release client session, endpoints={},
clientId={}", endpoints, clientId);
+ try {
+ requestObserver.onCompleted();
+ } catch (Throwable ignore) {
+ // Ignore exception on purpose.
+ }
}
}
void write(TelemetryCommand command) {
- if (null == requestObserver) {
- log.error("[Bug] Request observer does not exist, ignore current
command, endpoints={}, command={}, "
- + "clientId={}", endpoints, command,
sessionHandler.getClientId());
+ synchronized (requestObserverLock) {
+ if (null == requestObserver) {
+ log.error("[Bug] Request observer does not exist, ignore
current command, endpoints={}, command={}, "
+ + "clientId={}", endpoints, command,
sessionHandler.getClientId());
+ return;
+ }
+ requestObserver.onNext(command);
+ }
+ }
+
+ void reconnect() {
+ final ClientId clientId = sessionHandler.getClientId();
+ if (!reconnecting.compareAndSet(false, true)) {
+ log.info("Telemetry is already reconnecting, endpoints={},
clientId={}", endpoints, clientId);
return;
}
- requestObserver.onNext(command);
+ synchronized (requestObserverLock) {
+ if (null == requestObserver) {
+ reconnecting.set(false);
+ log.error("[Bug] request observer does not exist, failed to
reconnect telemetry, endpoints={}, "
+ + "clientId={}", endpoints, clientId);
+ return;
+ }
+ try {
+ requestObserver.onError(Status.CANCELLED.withDescription(
+ "Reconnect telemetry on transport
recovery").asRuntimeException());
+ } catch (Throwable t) {
+ reconnecting.set(false);
+ log.warn("Failed to cancel telemetry while reconnecting,
endpoints={}, clientId={}",
+ endpoints, clientId, t);
+ }
+ }
}
@Override
@@ -187,10 +226,21 @@ public class ClientSessionImpl implements
StreamObserver<TelemetryCommand> {
@Override
public void onError(Throwable throwable) {
final ClientId clientId = sessionHandler.getClientId();
- log.error("Exception raised from stream response observer,
clientId={}, endpoints={}", clientId, endpoints,
- throwable);
- release();
+ if (reconnecting.get()) {
+ if (Status.Code.CANCELLED ==
Status.fromThrowable(throwable).getCode()) {
+ log.info("Telemetry stream is cancelled for reconnecting,
clientId={}, endpoints={}",
+ clientId, endpoints);
+ } else {
+ log.warn("Telemetry stream failed while reconnecting,
clientId={}, endpoints={}",
+ clientId, endpoints, throwable);
+ }
+ } else {
+ log.error("Exception raised from stream response observer,
clientId={}, endpoints={}", clientId, endpoints,
+ throwable);
+ release();
+ }
if (!sessionHandler.isRunning()) {
+ reconnecting.set(false);
// first time to sync settings, forward the exception to upper
layer
settingsInitFuture.setException(throwable);
log.info("Session handler is not running, forgive to renew request
observer, clientId={}, "
@@ -205,8 +255,11 @@ public class ClientSessionImpl implements
StreamObserver<TelemetryCommand> {
public void onCompleted() {
final ClientId clientId = sessionHandler.getClientId();
log.info("Receive completion for stream response observer,
clientId={}, endpoints={}", clientId, endpoints);
- release();
+ if (!reconnecting.get()) {
+ release();
+ }
if (!sessionHandler.isRunning()) {
+ reconnecting.set(false);
log.info("Session handler is not running, forgive to renew request
observer, clientId={}, "
+ "endpoints={}", clientId, endpoints);
return;
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClient.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClient.java
index 98f252a5..9aaac5a5 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClient.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClient.java
@@ -43,6 +43,7 @@ import apache.rocketmq.v2.SyncLiteSubscriptionRequest;
import apache.rocketmq.v2.SyncLiteSubscriptionResponse;
import apache.rocketmq.v2.TelemetryCommand;
import com.google.common.util.concurrent.ListenableFuture;
+import io.grpc.ConnectivityState;
import io.grpc.Metadata;
import io.grpc.stub.StreamObserver;
import java.time.Duration;
@@ -69,6 +70,19 @@ public interface RpcClient {
*/
void shutdown() throws InterruptedException;
+ /**
+ * Move the channel into idle mode so that new RPCs create a new transport.
+ */
+ void enterIdle();
+
+ /**
+ * Get the current connectivity state.
+ *
+ * @param requestConnection if true, the channel will try to connect if it
is currently idle.
+ * @return current connectivity state.
+ */
+ ConnectivityState getState(boolean requestConnection);
+
/**
* Query topic route asynchronously.
*
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClientImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClientImpl.java
index 1e0225a1..328ac582 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClientImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcClientImpl.java
@@ -46,6 +46,7 @@ import apache.rocketmq.v2.TelemetryCommand;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.ClientInterceptor;
+import io.grpc.ConnectivityState;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
@@ -122,6 +123,16 @@ public class RpcClientImpl implements RpcClient {
channel.shutdown().awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
}
+ @Override
+ public void enterIdle() {
+ channel.enterIdle();
+ }
+
+ @Override
+ public ConnectivityState getState(boolean requestConnection) {
+ return channel.getState(requestConnection);
+ }
+
@Override
public ListenableFuture<QueryRouteResponse> queryRoute(Metadata metadata,
QueryRouteRequest request, Executor executor, Duration duration) {
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcFuture.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcFuture.java
index 83904a13..1fc8fbb7 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcFuture.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/rpc/RpcFuture.java
@@ -17,12 +17,17 @@
package org.apache.rocketmq.client.java.rpc;
+import apache.rocketmq.v2.Code;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
@SuppressWarnings("NullableProblems")
public class RpcFuture<R, T> implements ListenableFuture<T> {
@@ -33,7 +38,7 @@ public class RpcFuture<R, T> implements ListenableFuture<T> {
public RpcFuture(Context context, R request, ListenableFuture<T>
responseFuture) {
this.request = request;
this.context = context;
- this.responseFuture = responseFuture;
+ this.responseFuture = normalizeTransportException(context,
responseFuture);
}
public RpcFuture(Throwable t) {
@@ -50,6 +55,25 @@ public class RpcFuture<R, T> implements ListenableFuture<T> {
return context;
}
+ private static <T> ListenableFuture<T> normalizeTransportException(Context
context,
+ ListenableFuture<T> responseFuture) {
+ if (null == responseFuture) {
+ return null;
+ }
+ return Futures.catchingAsync(responseFuture,
StatusRuntimeException.class, exception -> {
+ if (Status.Code.RESOURCE_EXHAUSTED !=
exception.getStatus().getCode()) {
+ return Futures.immediateFailedFuture(exception);
+ }
+ final String requestId = null == context ? null :
context.getRequestId();
+ final String description = null ==
exception.getStatus().getDescription()
+ ? exception.getMessage() :
exception.getStatus().getDescription();
+ final TooManyRequestsException tooManyRequestsException = new
TooManyRequestsException(
+ Code.TOO_MANY_REQUESTS.getNumber(), requestId, description);
+ tooManyRequestsException.initCause(exception);
+ return Futures.immediateFailedFuture(tooManyRequestsException);
+ }, MoreExecutors.directExecutor());
+ }
+
@Override
public void addListener(Runnable listener, Executor executor) {
responseFuture.addListener(listener, executor);
diff --git
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientImplTest.java
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientImplTest.java
index 5f7ea7a8..ec1d8c00 100644
---
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientImplTest.java
+++
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientImplTest.java
@@ -33,6 +33,7 @@ import apache.rocketmq.v2.MessageType;
import apache.rocketmq.v2.NotifyClientTerminationRequest;
import apache.rocketmq.v2.Permission;
import apache.rocketmq.v2.PrintThreadStackTraceCommand;
+import apache.rocketmq.v2.ReconnectEndpointsCommand;
import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.TelemetryCommand;
import apache.rocketmq.v2.VerifyMessageCommand;
@@ -117,6 +118,19 @@ public class ClientImplTest extends TestBase {
verify(observer, times(1)).onNext(any(TelemetryCommand.class));
}
+ @Test
+ public void testOnReconnectEndpointsCommand() {
+ final Endpoints endpoints = fakeEndpoints();
+ final ReconnectEndpointsCommand command =
ReconnectEndpointsCommand.newBuilder().build();
+ final ClientManager clientManager = Mockito.mock(ClientManager.class);
+ final ClientImpl client = createClient();
+ doReturn(clientManager).when(client).getClientManager();
+
+ client.onReconnectEndpointsCommand(endpoints, command);
+
+ verify(clientManager, times(1)).reconnect(eq(endpoints));
+ }
+
@Test
public void testOnTopicRouteDataFetchedFailure() throws ClientException {
String topic = FAKE_TOPIC_0;
@@ -158,4 +172,4 @@ public class ClientImplTest extends TestBase {
doReturn(false).when(client).isRunning();
client.checkRunning();
}
-}
\ No newline at end of file
+}
diff --git
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientManagerImplTest.java
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientManagerImplTest.java
index a36c60c7..1dbdb0c3 100644
---
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientManagerImplTest.java
+++
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientManagerImplTest.java
@@ -22,6 +22,7 @@ import apache.rocketmq.v2.ChangeInvisibleDurationRequest;
import apache.rocketmq.v2.EndTransactionRequest;
import apache.rocketmq.v2.ForwardMessageToDeadLetterQueueRequest;
import apache.rocketmq.v2.HeartbeatRequest;
+import apache.rocketmq.v2.HeartbeatResponse;
import apache.rocketmq.v2.NotifyClientTerminationRequest;
import apache.rocketmq.v2.QueryAssignmentRequest;
import apache.rocketmq.v2.QueryRouteRequest;
@@ -29,11 +30,21 @@ import apache.rocketmq.v2.RecallMessageRequest;
import apache.rocketmq.v2.ReceiveMessageRequest;
import apache.rocketmq.v2.SendMessageRequest;
import apache.rocketmq.v2.SyncLiteSubscriptionRequest;
+import com.google.common.util.concurrent.Futures;
+import io.grpc.ConnectivityState;
import io.grpc.Metadata;
+import io.grpc.Status;
import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.java.misc.ClientId;
+import org.apache.rocketmq.client.java.route.Endpoints;
+import org.apache.rocketmq.client.java.rpc.RpcClient;
import org.apache.rocketmq.client.java.tool.TestBase;
import org.junit.AfterClass;
+import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
@@ -154,4 +165,158 @@ public class ClientManagerImplTest extends TestBase {
// Expect no exception thrown.
}
-}
\ No newline at end of file
+ @Test
+ public void testHeartbeatDeadlineExceededTriggersRecoveryAfterThreshold() {
+ final ClientManagerImpl clientManager = createClientManager();
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+ Mockito.verify(rpcClient, Mockito.never()).enterIdle();
+
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+ Mockito.verify(rpcClient, Mockito.times(1)).enterIdle();
+ }
+
+ @Test
+ public void testHeartbeatSuccessResetsFailureAttempts() {
+ final ClientManagerImpl clientManager = createClientManager();
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+ Futures.immediateFuture(HeartbeatResponse.getDefaultInstance()));
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+
+ Mockito.verify(rpcClient, Mockito.never()).enterIdle();
+ }
+
+ @Test
+ public void testHeartbeatUnavailableTriggersRecoveryImmediately() {
+ final ClientManagerImpl clientManager = createClientManager();
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
Mockito.when(rpcClient.getState(false)).thenReturn(ConnectivityState.READY);
+
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.UNAVAILABLE.asRuntimeException()));
+
+ Mockito.verify(rpcClient, Mockito.times(1)).enterIdle();
+ }
+
+ @Test
+ public void testHeartbeatUnavailableDoesNotRecoverNonReadyChannel() {
+ final ClientManagerImpl clientManager = createClientManager();
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
Mockito.when(rpcClient.getState(false)).thenReturn(ConnectivityState.TRANSIENT_FAILURE);
+
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.UNAVAILABLE.asRuntimeException()));
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+
+ Mockito.verify(rpcClient, Mockito.never()).enterIdle();
+ }
+
+ @Test
+ public void testHeartbeatResourceExhaustedDoesNotTriggerRecovery() {
+ final ClientManagerImpl clientManager = createClientManager();
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.RESOURCE_EXHAUSTED.asRuntimeException()));
+
+ Mockito.verify(rpcClient, Mockito.never()).enterIdle();
+ }
+
+ @Test
+ public void testHeartbeatRecoveryHasCooldown() {
+ final ClientManagerImpl clientManager = createClientManager();
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
+ for (int i = 0; i < 2 *
ClientManagerImpl.HEART_BEAT_FAILURE_THRESHOLD; i++) {
+ clientManager.monitorHeartbeat(fakeEndpoints(), rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+ }
+
+ Mockito.verify(rpcClient, Mockito.times(1)).enterIdle();
+ }
+
+ @Test
+ public void testServerReconnectIgnoresHeartbeatCooldown() {
+ final ClientManagerImpl clientManager = createClientManager();
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+ final Endpoints endpoints = fakeEndpoints();
+
+ for (int i = 0; i < ClientManagerImpl.HEART_BEAT_FAILURE_THRESHOLD;
i++) {
+ clientManager.monitorHeartbeat(endpoints, rpcClient,
+
Futures.immediateFailedFuture(Status.DEADLINE_EXCEEDED.asRuntimeException()));
+ }
+ clientManager.reconnect(endpoints, rpcClient);
+
+ Mockito.verify(rpcClient, Mockito.times(2)).enterIdle();
+ }
+
+ @Test
+ public void testConcurrentReconnectOnlyRecoversOnce() throws
InterruptedException {
+ final Client client = Mockito.mock(Client.class);
+ Mockito.when(client.getClientId()).thenReturn(FAKE_CLIENT_ID);
+ final ClientManagerImpl clientManager = new ClientManagerImpl(client);
+ final RpcClient rpcClient = Mockito.mock(RpcClient.class);
+ final Endpoints endpoints = fakeEndpoints();
+ final int threadCount = 8;
+ final CountDownLatch ready = new CountDownLatch(threadCount);
+ final CountDownLatch start = new CountDownLatch(1);
+ final CountDownLatch recoveryStarted = new CountDownLatch(1);
+ final CountDownLatch allowRecoveryToComplete = new CountDownLatch(1);
+ final CountDownLatch done = new CountDownLatch(threadCount);
+ final ExecutorService executor =
Executors.newFixedThreadPool(threadCount);
+ Mockito.doAnswer(invocation -> {
+ recoveryStarted.countDown();
+ Assert.assertTrue(allowRecoveryToComplete.await(5,
TimeUnit.SECONDS));
+ return null;
+ }).when(rpcClient).enterIdle();
+ try {
+ for (int i = 0; i < threadCount; i++) {
+ executor.execute(() -> {
+ ready.countDown();
+ try {
+ start.await();
+ clientManager.reconnect(endpoints, rpcClient);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ Assert.assertTrue(ready.await(5, TimeUnit.SECONDS));
+ start.countDown();
+ Assert.assertTrue(recoveryStarted.await(5, TimeUnit.SECONDS));
+ final long waitDeadlineNanos = System.nanoTime() +
TimeUnit.SECONDS.toNanos(5);
+ while (done.getCount() > 1 && System.nanoTime() <
waitDeadlineNanos) {
+ Thread.yield();
+ }
+ Assert.assertEquals(1, done.getCount());
+ allowRecoveryToComplete.countDown();
+ Assert.assertTrue(done.await(5, TimeUnit.SECONDS));
+ } finally {
+ allowRecoveryToComplete.countDown();
+ executor.shutdownNow();
+ }
+
+ Mockito.verify(rpcClient, Mockito.times(1)).enterIdle();
+ Mockito.verify(client,
Mockito.times(1)).reconnectTelemetry(Mockito.eq(endpoints));
+ }
+
+ private ClientManagerImpl createClientManager() {
+ final Client client = Mockito.mock(Client.class);
+ Mockito.when(client.getClientId()).thenReturn(FAKE_CLIENT_ID);
+ return new ClientManagerImpl(client);
+ }
+
+}
diff --git
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientSessionImplTest.java
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientSessionImplTest.java
index 0b2e4bf3..5fcfa3af 100644
---
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientSessionImplTest.java
+++
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/ClientSessionImplTest.java
@@ -25,13 +25,18 @@ import static org.mockito.Mockito.times;
import apache.rocketmq.v2.NotifyUnsubscribeLiteCommand;
import apache.rocketmq.v2.PrintThreadStackTraceCommand;
+import apache.rocketmq.v2.ReconnectEndpointsCommand;
import apache.rocketmq.v2.RecoverOrphanedTransactionCommand;
import apache.rocketmq.v2.Settings;
import apache.rocketmq.v2.TelemetryCommand;
import apache.rocketmq.v2.VerifyMessageCommand;
+import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
@@ -41,6 +46,7 @@ import
org.apache.rocketmq.client.java.impl.producer.ClientSessionHandler;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.apache.rocketmq.client.java.tool.TestBase;
import org.awaitility.Durations;
+import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
@@ -265,4 +271,137 @@ public class ClientSessionImplTest extends TestBase {
Mockito.verify(sessionHandler,
times(1)).onNotifyUnsubscribeLiteCommand(eq(endpoints), eq(command0));
}
-}
\ No newline at end of file
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testConcurrentReconnectOnlyCancelsTelemetryOnce() throws
ClientException, InterruptedException {
+ final Endpoints endpoints = fakeEndpoints();
+ final ClientSessionHandler sessionHandler =
Mockito.mock(ClientSessionHandler.class);
+ Mockito.when(sessionHandler.getScheduler()).thenReturn(new
ScheduledThreadPoolExecutor(1));
+ final StreamObserver<TelemetryCommand> requestObserver =
Mockito.mock(StreamObserver.class);
+
Mockito.doReturn(requestObserver).when(sessionHandler).telemetry(any(Endpoints.class),
+ any(StreamObserver.class));
+ Mockito.doReturn(FAKE_CLIENT_ID).when(sessionHandler).getClientId();
+ final ClientSessionImpl clientSession = new
ClientSessionImpl(sessionHandler, Duration.ofSeconds(3), endpoints);
+ final int threadCount = 8;
+ final CountDownLatch ready = new CountDownLatch(threadCount);
+ final CountDownLatch start = new CountDownLatch(1);
+ final CountDownLatch done = new CountDownLatch(threadCount);
+ final ExecutorService executor =
Executors.newFixedThreadPool(threadCount);
+ try {
+ for (int i = 0; i < threadCount; i++) {
+ executor.execute(() -> {
+ ready.countDown();
+ try {
+ start.await();
+ clientSession.reconnect();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ Assert.assertTrue(ready.await(5, TimeUnit.SECONDS));
+ start.countDown();
+ Assert.assertTrue(done.await(5, TimeUnit.SECONDS));
+ } finally {
+ executor.shutdownNow();
+ }
+
+ Mockito.verify(requestObserver,
times(1)).onError(any(Throwable.class));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testReconnectRenewsTelemetry() throws ClientException {
+ final Endpoints endpoints = fakeEndpoints();
+ final ClientSessionHandler sessionHandler =
Mockito.mock(ClientSessionHandler.class);
+ Mockito.doReturn(SCHEDULER).when(sessionHandler).getScheduler();
+ final StreamObserver<TelemetryCommand> firstRequestObserver =
Mockito.mock(StreamObserver.class);
+ final StreamObserver<TelemetryCommand> secondRequestObserver =
Mockito.mock(StreamObserver.class);
+ Mockito.doReturn(firstRequestObserver,
secondRequestObserver).when(sessionHandler).telemetry(
+ any(Endpoints.class), any(StreamObserver.class));
+ Mockito.doReturn(FAKE_CLIENT_ID).when(sessionHandler).getClientId();
+ Mockito.doReturn(true).when(sessionHandler).isRunning();
+
Mockito.doReturn(false).when(sessionHandler).isEndpointsDeprecated(endpoints);
+
Mockito.doReturn(TelemetryCommand.getDefaultInstance()).when(sessionHandler).settingsCommand();
+ final ClientSessionImpl clientSession = new
ClientSessionImpl(sessionHandler, Duration.ofSeconds(3), endpoints);
+
+ clientSession.reconnect();
+ clientSession.onError(Status.CANCELLED.asRuntimeException());
+
+
await().atMost(ClientSessionImpl.REQUEST_OBSERVER_RENEW_BACKOFF_DELAY.plus(Durations.ONE_SECOND))
+ .untilAsserted(() -> {
+ Mockito.verify(sessionHandler,
times(2)).telemetry(eq(endpoints), eq(clientSession));
+ Mockito.verify(secondRequestObserver,
times(1)).onNext(TelemetryCommand.getDefaultInstance());
+ });
+ clientSession.reconnect();
+
+ Mockito.verify(firstRequestObserver,
times(1)).onError(any(Throwable.class));
+ Mockito.verify(secondRequestObserver,
times(1)).onError(any(Throwable.class));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testReconnectCanRetryAfterSessionHandlerStops() throws
ClientException {
+ final Endpoints endpoints = fakeEndpoints();
+ final ClientSessionHandler sessionHandler =
Mockito.mock(ClientSessionHandler.class);
+ Mockito.doReturn(SCHEDULER).when(sessionHandler).getScheduler();
+ final StreamObserver<TelemetryCommand> requestObserver =
Mockito.mock(StreamObserver.class);
+
Mockito.doReturn(requestObserver).when(sessionHandler).telemetry(any(Endpoints.class),
+ any(StreamObserver.class));
+ Mockito.doReturn(FAKE_CLIENT_ID).when(sessionHandler).getClientId();
+ Mockito.doReturn(false).when(sessionHandler).isRunning();
+ final ClientSessionImpl clientSession = new
ClientSessionImpl(sessionHandler, Duration.ofSeconds(3), endpoints);
+
+ clientSession.reconnect();
+ clientSession.onError(Status.CANCELLED.asRuntimeException());
+ clientSession.reconnect();
+
+ Mockito.verify(requestObserver,
times(2)).onError(any(Throwable.class));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testReconnectCanRetryAfterEndpointsAreDeprecated() throws
ClientException {
+ final Endpoints endpoints = fakeEndpoints();
+ final ClientSessionHandler sessionHandler =
Mockito.mock(ClientSessionHandler.class);
+ Mockito.doReturn(SCHEDULER).when(sessionHandler).getScheduler();
+ final StreamObserver<TelemetryCommand> requestObserver =
Mockito.mock(StreamObserver.class);
+
Mockito.doReturn(requestObserver).when(sessionHandler).telemetry(any(Endpoints.class),
+ any(StreamObserver.class));
+ Mockito.doReturn(FAKE_CLIENT_ID).when(sessionHandler).getClientId();
+ Mockito.doReturn(true).when(sessionHandler).isRunning();
+
Mockito.doReturn(true).when(sessionHandler).isEndpointsDeprecated(endpoints);
+ final ClientSessionImpl clientSession = new
ClientSessionImpl(sessionHandler, Duration.ofSeconds(3), endpoints);
+
+ clientSession.reconnect();
+ clientSession.onError(Status.CANCELLED.asRuntimeException());
+
+
await().atMost(ClientSessionImpl.REQUEST_OBSERVER_RENEW_BACKOFF_DELAY.plus(Durations.ONE_SECOND))
+ .untilAsserted(() -> Mockito.verify(sessionHandler, times(1))
+ .removeClientSession(eq(endpoints), eq(clientSession)));
+ clientSession.reconnect();
+
+ Mockito.verify(requestObserver,
times(2)).onError(any(Throwable.class));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testOnNextWithReconnectEndpointsCommand() throws
ClientException {
+ final Endpoints endpoints = fakeEndpoints();
+ final ClientSessionHandler sessionHandler =
Mockito.mock(ClientSessionHandler.class);
+ Mockito.doReturn(SCHEDULER).when(sessionHandler).getScheduler();
+ final StreamObserver<TelemetryCommand> requestObserver =
Mockito.mock(StreamObserver.class);
+
Mockito.doReturn(requestObserver).when(sessionHandler).telemetry(any(Endpoints.class),
+ any(StreamObserver.class));
+ Mockito.doReturn(FAKE_CLIENT_ID).when(sessionHandler).getClientId();
+ final ClientSessionImpl clientSession = new
ClientSessionImpl(sessionHandler, Duration.ofSeconds(3), endpoints);
+ final ReconnectEndpointsCommand reconnectCommand =
ReconnectEndpointsCommand.getDefaultInstance();
+
+
clientSession.onNext(TelemetryCommand.newBuilder().setReconnectEndpointsCommand(reconnectCommand).build());
+
+ Mockito.verify(sessionHandler,
times(1)).onReconnectEndpointsCommand(eq(endpoints), eq(reconnectCommand));
+ }
+
+}
diff --git
a/java/client/src/test/java/org/apache/rocketmq/client/java/rpc/RpcFutureTest.java
b/java/client/src/test/java/org/apache/rocketmq/client/java/rpc/RpcFutureTest.java
new file mode 100644
index 00000000..0d0e8320
--- /dev/null
+++
b/java/client/src/test/java/org/apache/rocketmq/client/java/rpc/RpcFutureTest.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.client.java.rpc;
+
+import com.google.common.util.concurrent.Futures;
+import io.grpc.Metadata;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import java.util.concurrent.ExecutionException;
+import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RpcFutureTest {
+ private static final String REQUEST_ID = "fake-request-id";
+
+ private Context createContext() {
+ final Metadata metadata = new Metadata();
+ metadata.put(Metadata.Key.of(Signature.REQUEST_ID_KEY,
Metadata.ASCII_STRING_MARSHALLER), REQUEST_ID);
+ return new Context(null, metadata);
+ }
+
+ @Test
+ public void testResourceExhaustedIsNormalized() throws Exception {
+ final StatusRuntimeException statusRuntimeException =
+ Status.RESOURCE_EXHAUSTED.withDescription("flow
controlled").asRuntimeException();
+ final RpcFuture<Object, Object> future = new
RpcFuture<>(createContext(), null,
+ Futures.immediateFailedFuture(statusRuntimeException));
+
+ try {
+ future.get();
+ Assert.fail();
+ } catch (ExecutionException e) {
+ Assert.assertTrue(e.getCause() instanceof
TooManyRequestsException);
+
Assert.assertTrue(e.getCause().getMessage().contains("response-code=42900"));
+ Assert.assertTrue(e.getCause().getMessage().contains("request-id="
+ REQUEST_ID));
+ Assert.assertTrue(e.getCause().getMessage().contains("flow
controlled"));
+ Assert.assertSame(statusRuntimeException, e.getCause().getCause());
+ }
+ }
+
+ @Test
+ public void testOtherTransportExceptionIsUnchanged() throws Exception {
+ final StatusRuntimeException statusRuntimeException =
Status.UNAVAILABLE.asRuntimeException();
+ final RpcFuture<Object, Object> future = new
RpcFuture<>(createContext(), null,
+ Futures.immediateFailedFuture(statusRuntimeException));
+
+ try {
+ future.get();
+ Assert.fail();
+ } catch (ExecutionException e) {
+ Assert.assertSame(statusRuntimeException, e.getCause());
+ }
+ }
+
+ @Test
+ public void testSuccessfulResponseIsUnchanged() throws Exception {
+ final Object response = new Object();
+ final RpcFuture<Object, Object> future =
+ new RpcFuture<>(createContext(), null,
Futures.immediateFuture(response));
+
+ Assert.assertSame(response, future.get());
+ }
+}
diff --git
a/java/test/src/test/java/org/apache/rocketmq/test/client/ProducerHalfOpenTcpRecoveryIntegrationTest.java
b/java/test/src/test/java/org/apache/rocketmq/test/client/ProducerHalfOpenTcpRecoveryIntegrationTest.java
new file mode 100644
index 00000000..b8400f1c
--- /dev/null
+++
b/java/test/src/test/java/org/apache/rocketmq/test/client/ProducerHalfOpenTcpRecoveryIntegrationTest.java
@@ -0,0 +1,296 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.test.client;
+
+import apache.rocketmq.v2.Code;
+import apache.rocketmq.v2.SendMessageRequest;
+import apache.rocketmq.v2.SendMessageResponse;
+import apache.rocketmq.v2.SendResultEntry;
+import io.grpc.stub.StreamObserver;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.rocketmq.client.apis.ClientConfiguration;
+import org.apache.rocketmq.client.apis.ClientServiceProvider;
+import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
+import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
+import org.apache.rocketmq.client.apis.message.Message;
+import org.apache.rocketmq.client.apis.producer.Producer;
+import org.apache.rocketmq.client.java.message.MessageIdCodec;
+import org.apache.rocketmq.test.server.BaseMockServerImpl;
+import org.apache.rocketmq.test.server.GrpcServerIntegrationTest;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ProducerHalfOpenTcpRecoveryIntegrationTest extends
GrpcServerIntegrationTest {
+ private static final String LOOPBACK = "127.0.0.1";
+ private static final String TOPIC = "topic";
+ private static final Duration MAX_RECOVERY_TIME = Duration.ofSeconds(40);
+
+ private BaseMockServerImpl serverImpl;
+
+ @Before
+ public void setUp() throws Exception {
+ serverImpl = new ProducerMockServer(TOPIC);
+ setUpServer(serverImpl, port);
+ }
+
+ @Test(timeout = 60000)
+ public void testProducerRecoversFromHalfOpenTcpAfterHeartbeatTimeouts()
throws Exception {
+ final BlackholeTcpProxy proxy = new BlackholeTcpProxy(port);
+ serverImpl.setPort(proxy.getPort());
+ final ClientServiceProvider provider =
ClientServiceProvider.loadService();
+ final SessionCredentialsProvider credentials =
+ new StaticSessionCredentialsProvider("accessKey", "secretKey");
+ final ClientConfiguration clientConfiguration =
ClientConfiguration.newBuilder()
+ .setEndpoints(LOOPBACK + ":" + proxy.getPort())
+ .setCredentialProvider(credentials)
+ .setRequestTimeout(Duration.ofSeconds(3))
+ .build();
+ final Producer producer = provider.newProducerBuilder()
+ .setClientConfiguration(clientConfiguration)
+ .setTopics(TOPIC)
+ .setMaxAttempts(1)
+ .build();
+ final Message message = provider.newMessageBuilder()
+ .setTopic(TOPIC)
+ .setBody("tcp-blackhole".getBytes(StandardCharsets.UTF_8))
+ .build();
+
+ try {
+ producer.send(message);
+ proxy.assertHealthy();
+ final int initialConnectionCount =
proxy.getAcceptedConnectionCount();
+ Assert.assertTrue(initialConnectionCount > 0);
+ proxy.blackholeExistingConnections();
+ final long blackholeStartNanoTime = System.nanoTime();
+
+ assertSendFailure(producer, message);
+ Assert.assertEquals(0, proxy.getClosedConnectionCount());
+
+ while (Duration.ofNanos(System.nanoTime() -
blackholeStartNanoTime).compareTo(MAX_RECOVERY_TIME) < 0) {
+ proxy.assertHealthy();
+ try {
+ producer.send(message);
+ } catch (Exception ignore) {
+ // Keep sending until heartbeat recovery replaces the
blackholed TCP connection.
+ continue;
+ }
+ Assert.assertTrue(proxy.getAcceptedConnectionCount() >
initialConnectionCount);
+ return;
+ }
+ proxy.assertHealthy();
+ Assert.fail("Producer did not recover from the half-open TCP
connection within " + MAX_RECOVERY_TIME);
+ } finally {
+ producer.close();
+ proxy.close();
+ }
+ }
+
+ private static void assertSendFailure(Producer producer, Message message) {
+ Exception failure = null;
+ try {
+ producer.send(message);
+ } catch (Exception t) {
+ failure = t;
+ }
+ Assert.assertNotNull("Message should time out on the blackholed TCP
connection", failure);
+ }
+
+ private static final class ProducerMockServer extends BaseMockServerImpl {
+ private ProducerMockServer(String topic) {
+ super(topic);
+ }
+
+ @Override
+ public void sendMessage(SendMessageRequest request,
StreamObserver<SendMessageResponse> responseObserver) {
+ final apache.rocketmq.v2.Status status =
+
apache.rocketmq.v2.Status.newBuilder().setCode(Code.OK).build();
+ final SendResultEntry entry = SendResultEntry.newBuilder()
+ .setStatus(status)
+
.setMessageId(MessageIdCodec.getInstance().nextMessageId().toString())
+ .setOffset(1)
+ .build();
+ responseObserver.onNext(SendMessageResponse.newBuilder()
+ .setStatus(status)
+ .addEntries(entry)
+ .build());
+ responseObserver.onCompleted();
+ }
+ }
+
+ private static final class BlackholeTcpProxy implements AutoCloseable {
+ private final int backendPort;
+ private final ServerSocket serverSocket;
+ private final ExecutorService executor;
+ private final AtomicBoolean running;
+ private final AtomicInteger acceptedConnectionCount;
+ private final AtomicInteger closedConnectionCount;
+ private final AtomicInteger blackholeThroughConnectionId;
+ private final AtomicReference<IOException> acceptFailure;
+ private final List<SocketPair> connections;
+
+ private BlackholeTcpProxy(int backendPort) throws IOException {
+ this.backendPort = backendPort;
+ this.serverSocket = new ServerSocket();
+ this.serverSocket.bind(new
InetSocketAddress(InetAddress.getByName(LOOPBACK), 0));
+ this.executor = Executors.newCachedThreadPool(runnable -> {
+ final Thread thread = new Thread(runnable,
"BlackholeTcpProxy");
+ thread.setDaemon(true);
+ return thread;
+ });
+ this.running = new AtomicBoolean(true);
+ this.acceptedConnectionCount = new AtomicInteger();
+ this.closedConnectionCount = new AtomicInteger();
+ this.blackholeThroughConnectionId = new AtomicInteger();
+ this.acceptFailure = new AtomicReference<>();
+ this.connections = new CopyOnWriteArrayList<>();
+ this.executor.execute(this::acceptConnections);
+ }
+
+ private int getPort() {
+ return serverSocket.getLocalPort();
+ }
+
+ private int getAcceptedConnectionCount() {
+ return acceptedConnectionCount.get();
+ }
+
+ private int getClosedConnectionCount() {
+ return closedConnectionCount.get();
+ }
+
+ private void blackholeExistingConnections() {
+ blackholeThroughConnectionId.set(acceptedConnectionCount.get());
+ }
+
+ private void assertHealthy() {
+ final IOException failure = acceptFailure.get();
+ if (null == failure) {
+ return;
+ }
+ final AssertionError error = new AssertionError("TCP proxy stopped
accepting connections");
+ error.initCause(failure);
+ throw error;
+ }
+
+ private void acceptConnections() {
+ while (running.get()) {
+ try {
+ final Socket downstream = serverSocket.accept();
+ downstream.setTcpNoDelay(true);
+ final Socket upstream = new Socket();
+ upstream.setTcpNoDelay(true);
+ upstream.connect(new InetSocketAddress(LOOPBACK,
backendPort));
+ final int connectionId =
acceptedConnectionCount.incrementAndGet();
+ final SocketPair connection = new SocketPair(connectionId,
downstream, upstream);
+ connections.add(connection);
+ executor.execute(() -> forward(connection, downstream,
upstream));
+ executor.execute(() -> forward(connection, upstream,
downstream));
+ } catch (IOException e) {
+ if (running.get()) {
+ acceptFailure.compareAndSet(null, e);
+ return;
+ }
+ }
+ }
+ }
+
+ private void forward(SocketPair connection, Socket source, Socket
destination) {
+ final byte[] buffer = new byte[8192];
+ try {
+ final InputStream input = source.getInputStream();
+ final OutputStream output = destination.getOutputStream();
+ while (running.get()) {
+ final int length = input.read(buffer);
+ if (length < 0) {
+ break;
+ }
+ if (connection.id <= blackholeThroughConnectionId.get()) {
+ continue;
+ }
+ output.write(buffer, 0, length);
+ output.flush();
+ }
+ } catch (IOException ignore) {
+ // Socket closure is expected during transport recovery and
test cleanup.
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Override
+ public void close() {
+ if (!running.compareAndSet(true, false)) {
+ return;
+ }
+ try {
+ serverSocket.close();
+ } catch (IOException ignore) {
+ // Ignore exception on purpose.
+ }
+ connections.forEach(SocketPair::close);
+ executor.shutdownNow();
+ }
+
+ private final class SocketPair {
+ private final int id;
+ private final Socket downstream;
+ private final Socket upstream;
+ private final AtomicBoolean closed;
+
+ private SocketPair(int id, Socket downstream, Socket upstream) {
+ this.id = id;
+ this.downstream = downstream;
+ this.upstream = upstream;
+ this.closed = new AtomicBoolean();
+ }
+
+ private void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+ closeSocket(downstream);
+ closeSocket(upstream);
+ closedConnectionCount.incrementAndGet();
+ }
+
+ private void closeSocket(Socket socket) {
+ try {
+ socket.close();
+ } catch (IOException ignore) {
+ // Ignore exception on purpose.
+ }
+ }
+ }
+ }
+}
