opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4947] write client settings in lock and optimize the process of error handle (#4948)

Browse Source
This commit is contained in:
lk
2022-09-01 09:38:31 +08:00
committed by GitHub
parent f043ca8c18
commit 528b102a30
3 changed files with 132 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java
+60 -23
View File
@@ -30,6 +30,7 @@ import apache.rocketmq.v2.SubscriptionEntry;
import apache.rocketmq.v2.TelemetryCommand;
import apache.rocketmq.v2.ThreadStackTrace;
import apache.rocketmq.v2.VerifyMessageResult;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import java.util.HashSet;
import java.util.List;
@@ -42,6 +43,7 @@ import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.broker.client.ProducerGroupEvent;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.filter.FilterAPI;
@@ -179,7 +181,7 @@ public class ClientActivity extends AbstractMessingActivity {
try {
switch (request.getCommandCase()) {
case SETTINGS: {
responseObserver.onNext(processClientSettings(ctx, request, responseObserver));
processAndWriteClientSettings(ctx, request, responseObserver);
break;
}
case THREAD_STACK_TRACE: {
@@ -192,7 +194,7 @@ public class ClientActivity extends AbstractMessingActivity {
}
}
} catch (Throwable t) {
responseObserver.onNext(convertToTelemetryCommand(t));
processTelemetryException(request, t, responseObserver);
}
}
@@ -208,31 +210,63 @@ public class ClientActivity extends AbstractMessingActivity {
};
}
protected TelemetryCommand convertToTelemetryCommand(Throwable t) {
return TelemetryCommand.newBuilder().setStatus(ResponseBuilder.getInstance().buildStatus(t)).build();
}
protected TelemetryCommand processClientSettings(ProxyContext ctx, TelemetryCommand request,
StreamObserver<TelemetryCommand> responseObserver) {
String clientId = ctx.getClientID();
Settings settings = request.getSettings();
if (settings.hasPublishing()) {
for (Resource topic : settings.getPublishing().getTopicsList()) {
validateTopic(topic);
String topicName = GrpcConverter.getInstance().wrapResourceWithNamespace(topic);
GrpcClientChannel producerChannel = registerProducer(ctx, topicName);
producerChannel.setClientObserver(responseObserver);
protected void processTelemetryException(TelemetryCommand request, Throwable t, StreamObserver<TelemetryCommand> responseObserver) {
StatusRuntimeException exception = io.grpc.Status.INTERNAL
.withDescription("process client telemetryCommand failed. " + t.getMessage())
.withCause(t)
.asRuntimeException();
if (t instanceof GrpcProxyException) {
GrpcProxyException proxyException = (GrpcProxyException) t;
if (proxyException.getCode().getNumber() < Code.INTERNAL_ERROR_VALUE &&
proxyException.getCode().getNumber() >= Code.BAD_REQUEST_VALUE) {
exception = io.grpc.Status.INVALID_ARGUMENT
.withDescription("process client telemetryCommand failed. " + t.getMessage())
.withCause(t)
.asRuntimeException();
}
}
if (settings.hasSubscription()) {
validateConsumerGroup(settings.getSubscription().getGroup());
String groupName = GrpcConverter.getInstance().wrapResourceWithNamespace(settings.getSubscription().getGroup());
GrpcClientChannel consumerChannel = registerConsumer(ctx, groupName, settings.getClientType(), settings.getSubscription().getSubscriptionsList(), true);
consumerChannel.setClientObserver(responseObserver);
if (exception.getStatus().getCode().equals(io.grpc.Status.Code.INTERNAL)) {
log.warn("process client telemetryCommand failed. request:{}", request, t);
}
responseObserver.onError(exception);
}
protected void processAndWriteClientSettings(ProxyContext ctx, TelemetryCommand request,
StreamObserver<TelemetryCommand> responseObserver) {
GrpcClientChannel grpcClientChannel = null;
Settings settings = request.getSettings();
switch (settings.getPubSubCase()) {
case PUBLISHING:
for (Resource topic : settings.getPublishing().getTopicsList()) {
validateTopic(topic);
String topicName = GrpcConverter.getInstance().wrapResourceWithNamespace(topic);
grpcClientChannel = registerProducer(ctx, topicName);
grpcClientChannel.setClientObserver(responseObserver);
}
break;
case SUBSCRIPTION:
validateConsumerGroup(settings.getSubscription().getGroup());
String groupName = GrpcConverter.getInstance().wrapResourceWithNamespace(settings.getSubscription().getGroup());
grpcClientChannel = registerConsumer(ctx, groupName, settings.getClientType(), settings.getSubscription().getSubscriptionsList(), true);
grpcClientChannel.setClientObserver(responseObserver);
break;
default:
break;
}
if (grpcClientChannel == null) {
responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT
.withDescription("there is no publishing or subscription data in settings")
.asRuntimeException());
return;
}
TelemetryCommand command = processClientSettings(ctx, request);
grpcClientChannel.writeTelemetryCommand(command);
}
protected TelemetryCommand processClientSettings(ProxyContext ctx, TelemetryCommand request) {
String clientId = ctx.getClientID();
grpcClientSettingsManager.updateClientSettings(clientId, request.getSettings());
settings = grpcClientSettingsManager.getClientSettings(ctx);
Settings settings = grpcClientSettingsManager.getClientSettings(ctx);
return TelemetryCommand.newBuilder()
.setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
.setSettings(settings)
@@ -247,7 +281,10 @@ public class ClientActivity extends AbstractMessingActivity {
// use topic name as producer group
ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, parseClientVersion(ctx.getClientVersion()));
this.messagingProcessor.registerProducer(ctx, topicName, clientChannelInfo);
this.messagingProcessor.addTransactionSubscription(ctx, topicName, topicName);
TopicMessageType topicMessageType = this.messagingProcessor.getMetadataService().getTopicMessageType(topicName);
if (TopicMessageType.TRANSACTION.equals(topicMessageType)) {
this.messagingProcessor.addTransactionSubscription(ctx, topicName, topicName);
}
return channel;
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java
+4
View File
@@ -30,6 +30,7 @@ import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor;
import org.apache.rocketmq.proxy.service.metadata.MetadataService;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.junit.Ignore;
import org.junit.runner.RunWith;
@@ -47,6 +48,7 @@ public class BaseActivityTest extends InitConfigAndLoggerTest {
protected GrpcChannelManager grpcChannelManager;
protected ProxyRelayService proxyRelayService;
protected ReceiptHandleProcessor receiptHandleProcessor;
protected MetadataService metadataService;
protected static final String REMOTE_ADDR = "192.168.0.1:8080";
protected static final String LOCAL_ADDR = "127.0.0.1:8080";
@@ -61,12 +63,14 @@ public class BaseActivityTest extends InitConfigAndLoggerTest {
grpcClientSettingsManager = mock(GrpcClientSettingsManager.class);
proxyRelayService = mock(ProxyRelayService.class);
receiptHandleProcessor = mock(ReceiptHandleProcessor.class);
metadataService = mock(MetadataService.class);
metadata.put(InterceptorConstants.CLIENT_ID, CLIENT_ID);
metadata.put(InterceptorConstants.LANGUAGE, JAVA);
metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDR);
metadata.put(InterceptorConstants.LOCAL_ADDRESS, LOCAL_ADDR);
when(messagingProcessor.getProxyRelayService()).thenReturn(proxyRelayService);
when(messagingProcessor.getMetadataService()).thenReturn(metadataService);
grpcChannelManager = new GrpcChannelManager(messagingProcessor.getProxyRelayService());
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java
+68 -1
View File
@@ -33,10 +33,14 @@ import apache.rocketmq.v2.SubscriptionEntry;
import apache.rocketmq.v2.TelemetryCommand;
import apache.rocketmq.v2.ThreadStackTrace;
import apache.rocketmq.v2.VerifyMessageResult;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.CMResult;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
@@ -61,6 +65,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString;
@@ -129,6 +134,8 @@ public class ClientActivityTest extends BaseActivityTest {
txProducerTopicArgumentCaptor.capture()
);
when(this.metadataService.getTopicMessageType(anyString())).thenReturn(TopicMessageType.TRANSACTION);
HeartbeatResponse response = this.sendProducerHeartbeat(context);
assertEquals(Code.OK, response.getStatus().getCode());
@@ -147,6 +154,7 @@ public class ClientActivityTest extends BaseActivityTest {
Settings.newBuilder()
.setClientType(ClientType.PUSH_CONSUMER)
.setSubscription(Subscription.newBuilder()
.setGroup(Resource.newBuilder().setName("Group").build())
.addSubscriptions(SubscriptionEntry.newBuilder()
.setExpression(FilterExpression.newBuilder()
.setExpression("tag")
@@ -215,6 +223,7 @@ public class ClientActivityTest extends BaseActivityTest {
.build());
ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class);
doNothing().when(this.messagingProcessor).unRegisterProducer(any(), anyString(), channelInfoArgumentCaptor.capture());
when(this.metadataService.getTopicMessageType(anyString())).thenReturn(TopicMessageType.NORMAL);
this.sendProducerTelemetry(context);
this.sendProducerHeartbeat(context);
@@ -255,6 +264,64 @@ public class ClientActivityTest extends BaseActivityTest {
assertClientChannelInfo(clientChannelInfo, CONSUMER_GROUP);
}
@Test
public void testErrorConsumerGroupName() throws Throwable {
ProxyContext context = createContext();
try {
this.sendClientTelemetry(
context,
Settings.newBuilder()
.setClientType(ClientType.PUSH_CONSUMER)
.setSubscription(Subscription.newBuilder()
.addSubscriptions(SubscriptionEntry.newBuilder()
.setExpression(FilterExpression.newBuilder()
.setExpression("tag")
.setType(FilterType.TAG)
.build())
.setTopic(Resource.newBuilder().setName(TOPIC).build())
.build())
.build())
.build()).get();
fail();
} catch (ExecutionException e) {
StatusRuntimeException exception = (StatusRuntimeException) e.getCause();
assertEquals(Status.Code.INVALID_ARGUMENT, exception.getStatus().getCode());
}
}
@Test
public void testErrorProducerConfig() throws Throwable {
ProxyContext context = createContext();
try {
this.sendClientTelemetry(
context,
Settings.newBuilder()
.setClientType(ClientType.PRODUCER)
.setPublishing(Publishing.newBuilder()
.addTopics(Resource.newBuilder().setName("()").build())
.build())
.build()).get();
fail();
} catch (ExecutionException e) {
StatusRuntimeException exception = (StatusRuntimeException) e.getCause();
assertEquals(Status.Code.INVALID_ARGUMENT, exception.getStatus().getCode());
}
}
@Test
public void testEmptySettings() throws Throwable {
ProxyContext context = createContext();
try {
this.sendClientTelemetry(
context,
Settings.getDefaultInstance()).get();
fail();
} catch (ExecutionException e) {
StatusRuntimeException exception = (StatusRuntimeException) e.getCause();
assertEquals(Status.Code.INVALID_ARGUMENT, exception.getStatus().getCode());
}
}
@Test
public void testReportThreadStackTrace() {
this.clientActivity = new ClientActivity(this.messagingProcessor, this.grpcClientSettingsManager, grpcChannelManagerMock);
@@ -331,7 +398,7 @@ public class ClientActivityTest extends BaseActivityTest {
@Override
public void onError(Throwable t) {
future.completeExceptionally(t);
}
@Override