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] Implement Local transaction module

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:29:11 +08:00
parent 67440ebe27
commit 5cbb01e373
5 changed files with 130 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+8 -2
View File
@@ -185,6 +185,7 @@ public class BrokerController {
protected final Broker2Client broker2Client;
protected final SubscriptionGroupManager subscriptionGroupManager;
protected final ConsumerIdsChangeListener consumerIdsChangeListener;
protected final EndTransactionProcessor endTransactionProcessor;
private final RebalanceLockManager rebalanceLockManager = new RebalanceLockManager();
protected BrokerOuterAPI brokerOuterAPI;
protected ScheduledExecutorService scheduledExecutorService;
@@ -317,6 +318,7 @@ public class BrokerController {
this.queryAssignmentProcessor = new QueryAssignmentProcessor(this);
this.clientManageProcessor = new ClientManageProcessor(this);
this.slaveSynchronize = new SlaveSynchronize(this);
this.endTransactionProcessor = new EndTransactionProcessor(this);
this.sendThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getSendThreadPoolQueueCapacity());
this.putThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPutThreadPoolQueueCapacity());
@@ -1002,8 +1004,8 @@ public class BrokerController {
/**
* EndTransactionProcessor
*/
this.remotingServer.registerProcessor(RequestCode.END_TRANSACTION, new EndTransactionProcessor(this), this.endTransactionExecutor);
this.fastRemotingServer.registerProcessor(RequestCode.END_TRANSACTION, new EndTransactionProcessor(this), this.endTransactionExecutor);
this.remotingServer.registerProcessor(RequestCode.END_TRANSACTION, endTransactionProcessor, this.endTransactionExecutor);
this.fastRemotingServer.registerProcessor(RequestCode.END_TRANSACTION, endTransactionProcessor, this.endTransactionExecutor);
/*
* Default
@@ -2104,6 +2106,10 @@ public class BrokerController {
return brokerPreOnlineService;
}
public EndTransactionProcessor getEndTransactionProcessor() {
return endTransactionProcessor;
}
public boolean isScheduleServiceStart() {
return isScheduleServiceStart;
}
proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java
+3 -3
View File
@@ -91,8 +91,8 @@ public class TransactionId {
.build();
}
public static TransactionId genFromGatewayTransactionId(String gatewayTransactionId) throws UnknownHostException {
ByteBuffer byteBuffer = ByteBuffer.wrap(UtilAll.string2bytes(gatewayTransactionId));
public static TransactionId decode(String transactionId) throws UnknownHostException {
ByteBuffer byteBuffer = ByteBuffer.wrap(UtilAll.string2bytes(transactionId));
byte[] ip = new byte[4];
byteBuffer.get(ip);
@@ -111,7 +111,7 @@ public class TransactionId {
.brokerTransactionId(new String(orgTransactionIdByte, StandardCharsets.UTF_8))
.commitLogOffset(commitLogOffset)
.tranStateTableOffset(tranStateTableOffset)
.gatewayTransactionId(gatewayTransactionId)
.gatewayTransactionId(transactionId)
.build();
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/GrpcClientChannel.java
+39
View File
@@ -17,11 +17,19 @@
package org.apache.rocketmq.proxy.grpc.adapter.channel;
import apache.rocketmq.v1.PollCommandResponse;
import apache.rocketmq.v1.RecoverOrphanedTransactionCommand;
import io.netty.channel.ChannelFuture;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.proxy.channel.ChannelManager;
import org.apache.rocketmq.proxy.channel.SimpleChannel;
import org.apache.rocketmq.proxy.grpc.common.Converter;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class GrpcClientChannel extends SimpleChannel {
private final AtomicReference<CompletableFuture<PollCommandResponse>> pollCommandResponseFutureRef = new AtomicReference<>();
@@ -56,9 +64,40 @@ public class GrpcClientChannel extends SimpleChannel {
return group + "@" + clientId;
}
/**
* Write response to corresponding remote client
*
* @param msg Target write object, {@link RemotingCommand} or {@link PollCommandResponse}
* @return Always success {@link ChannelFuture}
* <p>
* Case {@link RequestCode#CHECK_TRANSACTION_STATE}
* @see org.apache.rocketmq.broker.client.net.Broker2Client#checkProducerTransactionState
*/
@Override
public ChannelFuture writeAndFlush(Object msg) {
CompletableFuture<PollCommandResponse> future = pollCommandResponseFutureRef.get();
if (msg instanceof RemotingCommand) {
RemotingCommand command = (RemotingCommand) msg;
try {
switch (command.getCode()) {
case RequestCode.CHECK_TRANSACTION_STATE: {
final CheckTransactionStateRequestHeader requestHeader =
(CheckTransactionStateRequestHeader) command.decodeCommandCustomHeader(CheckTransactionStateRequestHeader.class);
MessageExt messageExt = MessageDecoder.decode(ByteBuffer.wrap(command.getBody()), true, false, false);
RecoverOrphanedTransactionCommand recoverOrphanedTransactionCommand = RecoverOrphanedTransactionCommand.newBuilder()
.setTransactionId(requestHeader.getTransactionId())
.setOrphanedTransactionalMessage(Converter.buildMessage(messageExt))
.build();
future.complete(PollCommandResponse.newBuilder()
.setRecoverOrphanedTransactionCommand(recoverOrphanedTransactionCommand)
.build());
break;
}
}
} catch (Exception e) {
}
}
if (msg instanceof PollCommandResponse) {
PollCommandResponse response = (PollCommandResponse) msg;
future.complete(response);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/common/Converter.java
+42
View File
@@ -25,6 +25,7 @@ import apache.rocketmq.v1.ConsumerData;
import apache.rocketmq.v1.Digest;
import apache.rocketmq.v1.DigestType;
import apache.rocketmq.v1.Encoding;
import apache.rocketmq.v1.EndTransactionRequest;
import apache.rocketmq.v1.FilterExpression;
import apache.rocketmq.v1.FilterType;
import apache.rocketmq.v1.ForwardMessageToDeadLetterQueueRequest;
@@ -46,6 +47,7 @@ import com.google.protobuf.Timestamp;
import com.google.protobuf.util.Durations;
import com.google.protobuf.util.Timestamps;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
@@ -67,6 +69,7 @@ import org.apache.rocketmq.common.protocol.NamespaceUtil;
import org.apache.rocketmq.common.protocol.header.AckMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeRequestHeader;
import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
@@ -75,6 +78,7 @@ import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.utils.BinaryUtil;
import org.apache.rocketmq.proxy.connector.transaction.TransactionId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -198,6 +202,33 @@ public class Converter {
return consumerSendMsgBackRequestHeader;
}
public static EndTransactionRequestHeader buildEndTransactionRequestHeader(EndTransactionRequest request) {
String groupName = Converter.getResourceNameWithNamespace(request.getGroup());
String messageId = request.getMessageId();
String transactionId = request.getTransactionId();
TransactionId handle;
try {
handle = TransactionId.decode(transactionId);
} catch (UnknownHostException e) {
throw new IllegalArgumentException("Parse transaction id failed", e);
}
long transactionStateTableOffset = handle.getTranStateTableOffset();
long commitLogOffset = handle.getCommitLogOffset();
boolean fromTransactionCheck = request.getSource() == EndTransactionRequest.Source.SERVER_CHECK;
int commitOrRollback = Converter.buildTransactionCommitOrRollback(request.getResolution());
EndTransactionRequestHeader endTransactionRequestHeader = new EndTransactionRequestHeader();
endTransactionRequestHeader.setProducerGroup(groupName);
endTransactionRequestHeader.setMsgId(messageId);
endTransactionRequestHeader.setTransactionId(transactionId);
endTransactionRequestHeader.setTranStateTableOffset(transactionStateTableOffset);
endTransactionRequestHeader.setCommitLogOffset(commitLogOffset);
endTransactionRequestHeader.setCommitOrRollback(commitOrRollback);
endTransactionRequestHeader.setFromTransactionCheck(fromTransactionCheck);
return endTransactionRequestHeader;
}
public static Map<String, String> buildMessageProperty(Message message) {
org.apache.rocketmq.common.message.Message messageWithHeader = new org.apache.rocketmq.common.message.Message();
// set user properties
@@ -546,4 +577,15 @@ public class Converter {
return systemAttributeBuilder.build();
}
public static int buildTransactionCommitOrRollback(EndTransactionRequest.TransactionResolution type) {
switch (type) {
case COMMIT:
return MessageSysFlag.TRANSACTION_COMMIT_TYPE;
case ROLLBACK:
return MessageSysFlag.TRANSACTION_ROLLBACK_TYPE;
default:
return MessageSysFlag.TRANSACTION_NOT_TYPE;
}
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java
+38 -1
View File
@@ -32,6 +32,7 @@ import apache.rocketmq.v1.HeartbeatResponse;
import apache.rocketmq.v1.Message;
import apache.rocketmq.v1.NackMessageRequest;
import apache.rocketmq.v1.NackMessageResponse;
import apache.rocketmq.v1.NoopCommand;
import apache.rocketmq.v1.NotifyClientTerminationRequest;
import apache.rocketmq.v1.NotifyClientTerminationResponse;
import apache.rocketmq.v1.PollCommandRequest;
@@ -68,6 +69,7 @@ import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.header.AckMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeRequestHeader;
import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
@@ -304,7 +306,30 @@ public class LocalGrpcService implements GrpcForwardService {
@Override
public CompletableFuture<EndTransactionResponse> endTransaction(Context ctx, EndTransactionRequest request) {
return null;
Channel channel = channelManager.createChannel();
SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel);
EndTransactionRequestHeader requestHeader = Converter.buildEndTransactionRequestHeader(request);
RemotingCommand command = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, requestHeader);
command.makeCustomHeaderToNet();
CompletableFuture<EndTransactionResponse> future = new CompletableFuture<>();
try {
RemotingCommand responseCommand = brokerController.getEndTransactionProcessor()
.processRequest(channelHandlerContext, command);
EndTransactionResponse.Builder builder = EndTransactionResponse.newBuilder();
if (null != responseCommand) {
builder.setCommon(ResponseBuilder.buildCommon(responseCommand.getCode(), responseCommand.getRemark()));
} else {
builder.setCommon(ResponseBuilder.buildCommon(Code.INTERNAL, "Response command is null"));
}
EndTransactionResponse response = builder.build();
future.complete(response);
} catch (Exception e) {
LOGGER.error("Exception raised while endTransaction", e);
future.completeExceptionally(e);
}
return future;
}
@Override public CompletableFuture<QueryOffsetResponse> queryOffset(Context ctx, QueryOffsetRequest request) {
@@ -323,12 +348,24 @@ public class LocalGrpcService implements GrpcForwardService {
Resource producerGroup = request.getProducerGroup();
String producerGroupName = Converter.getResourceNameWithNamespace(producerGroup);
GrpcClientChannel producerChannel = GrpcClientChannel.getChannel(channelManager, producerGroupName, clientId);
if (producerChannel == null) {
future.complete(PollCommandResponse.newBuilder()
.setNoopCommand(NoopCommand.newBuilder().build())
.build());
break;
}
producerChannel.addClientObserver(future);
break;
case CONSUMER_GROUP:
Resource consumerGroup = request.getConsumerGroup();
String consumerGroupName = Converter.getResourceNameWithNamespace(consumerGroup);
GrpcClientChannel consumerChannel = GrpcClientChannel.getChannel(channelManager, consumerGroupName, clientId);
if (consumerChannel == null) {
future.complete(PollCommandResponse.newBuilder()
.setNoopCommand(NoopCommand.newBuilder().build())
.build());
break;
}
consumerChannel.addClientObserver(future);
break;
default: