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

[ISSUE #3949] Do some refactoring work.

Browse Source
This commit is contained in:
Jixiang.jjx
2022-07-13 11:29:15 +08:00
committed by zhouxiang
parent e91ccf0bd6
commit ff3117f322
15 changed files with 39 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/connector/AbstractForwardClient.java
+2 -1
View File
@@ -46,7 +46,8 @@ public abstract class AbstractForwardClient implements StartAndShutdown {
if (clients.length == 1) {
return this.clients[0];
}
return this.clients[ThreadLocalRandom.current().nextInt(this.clients.length)];
int index = ThreadLocalRandom.current().nextInt(this.clients.length);
return this.clients[index];
}
@Override
proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardProducer.java
+1 -1
View File
@@ -87,7 +87,7 @@ public class ForwardProducer extends AbstractForwardClient {
return future.thenApply(sendResult -> {
int tranType = MessageSysFlag.getTransactionValue(requestHeader.getSysFlag());
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) && tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE) {
TransactionId transactionId = TransactionId.genFromBrokerTransactionId(address, sendResult);
TransactionId transactionId = TransactionId.genByBrokerTransactionId(address, sendResult);
sendResult.setTransactionId(transactionId.getProxyTransactionId());
}
return sendResult;
proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/ForwardClientManager.java
+1 -1
View File
@@ -30,7 +30,7 @@ import org.apache.rocketmq.remoting.RPCHook;
public class ForwardClientManager implements StartAndShutdown {
private RPCHook rpcHook = null;
private RPCHook rpcHook;
private final MQClientFactory mqClientFactory;
private final TransactionProducerFactory transactionalProducerFactory;
proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/MQClientFactory.java
+1 -2
View File
@@ -23,8 +23,7 @@ import org.apache.rocketmq.remoting.RPCHook;
public class MQClientFactory extends AbstractMQClientFactory {
public MQClientFactory(ScheduledExecutorService scheduledExecutorService,
RPCHook rpcHook) {
public MQClientFactory(ScheduledExecutorService scheduledExecutorService, RPCHook rpcHook) {
super(scheduledExecutorService, rpcHook);
}
proxy/src/main/java/org/apache/rocketmq/proxy/connector/processor/ProxyClientRemotingProcessor.java
+1 -1
View File
@@ -63,7 +63,7 @@ public class ProxyClientRemotingProcessor extends ClientRemotingProcessor {
requestHeader.getTranStateTableOffset(),
requestHeader.getCommitLogOffset(),
requestHeader.getMsgId(),
TransactionId.genFromBrokerTransactionId(
TransactionId.genByBrokerTransactionId(
ctx.channel().remoteAddress(),
requestHeader.getTransactionId(),
requestHeader.getCommitLogOffset(),
proxy/src/main/java/org/apache/rocketmq/proxy/connector/route/TopicRouteCache.java
+2 -1
View File
@@ -81,7 +81,8 @@ public class TopicRouteCache {
}
public SelectableMessageQueue selectOneWriteQueue(String topic, String brokerName, int queueId) throws Exception {
return getMessageQueue(topic).getWriteSelector().selectOne(brokerName, queueId);
return getMessageQueue(topic).getWriteSelector()
.selectOne(brokerName, queueId);
}
public SelectableMessageQueue selectOneWriteQueueByKey(String topic, String shardingKey) throws Exception {
proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionHeartbeatRegisterService.java
-1
View File
@@ -40,7 +40,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TransactionHeartbeatRegisterService implements StartAndShutdown {
private static final Logger log = LoggerFactory.getLogger(TransactionHeartbeatRegisterService.class);
private static final String TRANS_HEARTBEAT_CLIENT_ID = "rmq-proxy-producer-client";
proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java
+3 -3
View File
@@ -54,7 +54,7 @@ public class TransactionId {
public TransactionId() {
}
public static TransactionId genFromBrokerTransactionId(String brokerAddr, SendResult sendResult) {
public static TransactionId genByBrokerTransactionId(String brokerAddr, SendResult sendResult) {
MessageId id = new MessageId(null, 0);
try {
if (sendResult.getOffsetMsgId() != null) {
@@ -65,11 +65,11 @@ public class TransactionId {
} catch (Exception e) {
log.warn("genFromBrokerTransactionId failed. brokerAddr: {}, sendResult: {}", brokerAddr, sendResult, e);
}
return genFromBrokerTransactionId(RemotingUtil.string2SocketAddress(brokerAddr), sendResult.getTransactionId(),
return genByBrokerTransactionId(RemotingUtil.string2SocketAddress(brokerAddr), sendResult.getTransactionId(),
id.getOffset(), sendResult.getQueueOffset());
}
public static TransactionId genFromBrokerTransactionId(SocketAddress brokerAddr, String orgTransactionId,
public static TransactionId genByBrokerTransactionId(SocketAddress brokerAddr, String orgTransactionId,
long commitLogOffset, long tranStateTableOffset) {
byte[] orgTransactionIdByte = new byte[0];
if (StringUtils.isNotBlank(orgTransactionId)) {
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcMessagingProcessor.java
+10 -5
View File
@@ -189,7 +189,8 @@ public class GrpcMessagingProcessor extends MessagingServiceGrpc.MessagingServic
}
@Override
public void forwardMessageToDeadLetterQueue(ForwardMessageToDeadLetterQueueRequest request, StreamObserver<ForwardMessageToDeadLetterQueueResponse> responseObserver) {
public void forwardMessageToDeadLetterQueue(ForwardMessageToDeadLetterQueueRequest request,
StreamObserver<ForwardMessageToDeadLetterQueueResponse> responseObserver) {
CompletableFuture<ForwardMessageToDeadLetterQueueResponse> future = grpcForwardService.forwardMessageToDeadLetterQueue(Context.current(), request);
future.thenAccept(response -> ResponseWriter.write(responseObserver, response))
.exceptionally(e -> {
@@ -251,7 +252,8 @@ public class GrpcMessagingProcessor extends MessagingServiceGrpc.MessagingServic
}
@Override
public void reportThreadStackTrace(ReportThreadStackTraceRequest request, StreamObserver<ReportThreadStackTraceResponse> responseObserver) {
public void reportThreadStackTrace(ReportThreadStackTraceRequest request,
StreamObserver<ReportThreadStackTraceResponse> responseObserver) {
CompletableFuture<ReportThreadStackTraceResponse> future = grpcForwardService.reportThreadStackTrace(Context.current(), request);
future.thenAccept(response -> ResponseWriter.write(responseObserver, response))
.exceptionally(e -> {
@@ -264,7 +266,8 @@ public class GrpcMessagingProcessor extends MessagingServiceGrpc.MessagingServic
}
@Override
public void reportMessageConsumptionResult(ReportMessageConsumptionResultRequest request, StreamObserver<ReportMessageConsumptionResultResponse> responseObserver) {
public void reportMessageConsumptionResult(ReportMessageConsumptionResultRequest request,
StreamObserver<ReportMessageConsumptionResultResponse> responseObserver) {
CompletableFuture<ReportMessageConsumptionResultResponse> future = grpcForwardService.reportMessageConsumptionResult(Context.current(), request);
future.thenAccept(response -> ResponseWriter.write(responseObserver, response))
.exceptionally(e -> {
@@ -277,7 +280,8 @@ public class GrpcMessagingProcessor extends MessagingServiceGrpc.MessagingServic
}
@Override
public void notifyClientTermination(NotifyClientTerminationRequest request, StreamObserver<NotifyClientTerminationResponse> responseObserver) {
public void notifyClientTermination(NotifyClientTerminationRequest request,
StreamObserver<NotifyClientTerminationResponse> responseObserver) {
CompletableFuture<NotifyClientTerminationResponse> future = grpcForwardService.notifyClientTermination(Context.current(), request);
future.thenAccept(response -> ResponseWriter.write(responseObserver, response))
.exceptionally(e -> {
@@ -290,7 +294,8 @@ public class GrpcMessagingProcessor extends MessagingServiceGrpc.MessagingServic
}
@Override
public void changeInvisibleDuration(ChangeInvisibleDurationRequest request, StreamObserver<ChangeInvisibleDurationResponse> responseObserver) {
public void changeInvisibleDuration(ChangeInvisibleDurationRequest request,
StreamObserver<ChangeInvisibleDurationResponse> responseObserver) {
CompletableFuture<ChangeInvisibleDurationResponse> future = grpcForwardService.changeInvisibleDuration(Context.current(), request);
future.thenAccept(response -> ResponseWriter.write(responseObserver, response))
.exceptionally(e -> {
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java
+3 -2
View File
@@ -97,6 +97,7 @@ public class GrpcServer implements StartAndShutdown {
.addService(messagingProcessor)
.executor(this.executor);
// grpc interceptors, including acl, logging etc.
if (ConfigurationManager.getProxyConfig().isEnableACL()) {
List<AccessValidator> accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class);
if (accessValidators.isEmpty()) {
@@ -122,7 +123,7 @@ public class GrpcServer implements StartAndShutdown {
this.grpcForwardService.start();
this.server.start();
log.info("grpc server has started");
log.info("grpc server start successfully.");
}
public void shutdown() {
@@ -132,7 +133,7 @@ public class GrpcServer implements StartAndShutdown {
this.grpcForwardService.shutdown();
log.info("grpc server has stopped");
log.info("grpc server shutdown successfully.");
} catch (Exception e) {
e.printStackTrace();
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/GrpcConverter.java
+5 -10
View File
@@ -175,8 +175,7 @@ public class GrpcConverter {
public static AckMessageRequestHeader buildAckMessageRequestHeader(AckMessageRequest request) {
String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
String topicName = GrpcConverter.wrapResourceWithNamespace(request.getTopic());
String receiptHandleStr = request.getReceiptHandle();
ReceiptHandle handle = ReceiptHandle.decode(receiptHandleStr);
ReceiptHandle handle = ReceiptHandle.decode(request.getReceiptHandle());
AckMessageRequestHeader ackMessageRequestHeader = new AckMessageRequestHeader();
ackMessageRequestHeader.setConsumerGroup(groupName);
@@ -191,8 +190,7 @@ public class GrpcConverter {
DelayPolicy delayPolicy) {
String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
String topicName = GrpcConverter.wrapResourceWithNamespace(request.getTopic());
String receiptHandleStr = request.getReceiptHandle();
ReceiptHandle handle = ReceiptHandle.decode(receiptHandleStr);
ReceiptHandle handle = ReceiptHandle.decode(request.getReceiptHandle());
ChangeInvisibleTimeRequestHeader changeInvisibleTimeRequestHeader = new ChangeInvisibleTimeRequestHeader();
changeInvisibleTimeRequestHeader.setConsumerGroup(groupName);
@@ -209,8 +207,7 @@ public class GrpcConverter {
ChangeInvisibleDurationRequest request) {
String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
String topicName = GrpcConverter.wrapResourceWithNamespace(request.getTopic());
String receiptHandleStr = request.getReceiptHandle();
ReceiptHandle handle = ReceiptHandle.decode(receiptHandleStr);
ReceiptHandle handle = ReceiptHandle.decode(request.getReceiptHandle());
ChangeInvisibleTimeRequestHeader changeInvisibleTimeRequestHeader = new ChangeInvisibleTimeRequestHeader();
changeInvisibleTimeRequestHeader.setConsumerGroup(groupName);
@@ -226,8 +223,7 @@ public class GrpcConverter {
ForwardMessageToDeadLetterQueueRequest request) {
String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
String topicName = GrpcConverter.wrapResourceWithNamespace(request.getTopic());
String receiptHandleStr = request.getReceiptHandle();
ReceiptHandle handle = ReceiptHandle.decode(receiptHandleStr);
ReceiptHandle handle = ReceiptHandle.decode(request.getReceiptHandle());
ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader = new ConsumerSendMsgBackRequestHeader();
consumerSendMsgBackRequestHeader.setOffset(handle.getCommitLogOffset());
@@ -243,8 +239,7 @@ public class GrpcConverter {
NackMessageRequest request) {
String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
String topicName = GrpcConverter.wrapResourceWithNamespace(request.getTopic());
String receiptHandleStr = request.getReceiptHandle();
ReceiptHandle handle = ReceiptHandle.decode(receiptHandleStr);
ReceiptHandle handle = ReceiptHandle.decode(request.getReceiptHandle());
ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader = new ConsumerSendMsgBackRequestHeader();
consumerSendMsgBackRequestHeader.setOffset(handle.getCommitLogOffset());
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ForwardClientService.java
+4 -2
View File
@@ -68,10 +68,12 @@ public class ForwardClientService extends BaseService {
this.pollCommandResponseManager = pollCommandResponseManager;
this.consumerManager = new ConsumerManager(new ConsumerIdsChangeListener() {
@Override public void handle(ConsumerGroupEvent event, String group, Object... args) {
@Override
public void handle(ConsumerGroupEvent event, String group, Object... args) {
}
@Override public void shutdown() {
@Override
public void shutdown() {
}
});
this.producerManager = new ProducerManager();
proxy/src/test/java/org/apache/rocketmq/proxy/connector/transaction/TransactionIdTest.java
+3 -3
View File
@@ -10,7 +10,7 @@ public class TransactionIdTest {
@Test
public void test() throws UnknownHostException {
TransactionId transactionId = TransactionId.genFromBrokerTransactionId(
TransactionId transactionId = TransactionId.genByBrokerTransactionId(
RemotingHelper.string2SocketAddress("127.0.0.1:8080"),
"71F99B78B6E261357FA259CCA6456118", 1234, 5678);
@@ -24,7 +24,7 @@ public class TransactionIdTest {
@Test
public void testEmptyTransactionId() throws UnknownHostException {
TransactionId transactionId = TransactionId.genFromBrokerTransactionId(
TransactionId transactionId = TransactionId.genByBrokerTransactionId(
RemotingHelper.string2SocketAddress("127.0.0.1:8080"),
"", 1234, 5678);
@@ -38,7 +38,7 @@ public class TransactionIdTest {
@Test
public void testNullTransactionId() throws UnknownHostException {
TransactionId transactionId = TransactionId.genFromBrokerTransactionId(
TransactionId transactionId = TransactionId.genByBrokerTransactionId(
RemotingHelper.string2SocketAddress("127.0.0.1:8080"),
null, 1234, 5678);
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcServiceTest.java
+1 -1
View File
@@ -381,7 +381,7 @@ public class LocalGrpcServiceTest extends InitConfigAndLoggerTest {
.thenReturn(response);
EndTransactionRequest request = EndTransactionRequest.newBuilder()
.setMessageId("123")
.setTransactionId(TransactionId.genFromBrokerTransactionId(
.setTransactionId(TransactionId.genByBrokerTransactionId(
new InetSocketAddress("0.0.0.0", 80), "123", 123, 123
).getProxyTransactionId()
)
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/TransactionServiceTest.java
+2 -2
View File
@@ -48,7 +48,7 @@ public class TransactionServiceTest extends BaseServiceTest {
return null;
}).when(channel).writeAndFlush(any());
TransactionId transactionId = TransactionId.genFromBrokerTransactionId(
TransactionId transactionId = TransactionId.genByBrokerTransactionId(
RemotingHelper.string2SocketAddress("127.0.0.1:8080"),
"71F99B78B6E261357FA259CCA6456118", 1234, 5678);
transactionService.checkTransactionState(new TransactionStateCheckRequest(
@@ -69,7 +69,7 @@ public class TransactionServiceTest extends BaseServiceTest {
public void testEndTransaction() throws Exception {
AtomicReference<EndTransactionRequestHeader> headerRef = new AtomicReference<>();
AtomicReference<String> brokerAddrRef = new AtomicReference<>();
TransactionId transactionId = TransactionId.genFromBrokerTransactionId(
TransactionId transactionId = TransactionId.genByBrokerTransactionId(
RemotingHelper.string2SocketAddress("127.0.0.1:8080"),
"71F99B78B6E261357FA259CCA6456118", 1234, 5678);
doAnswer(mock -> {