diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java index 44a4e902dd..7bf9cdcd9c 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java @@ -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(this.brokerConfig.getSendThreadPoolQueueCapacity()); this.putThreadPoolQueue = new LinkedBlockingQueue(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; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java index 63eb8b3f08..c04cb82bb3 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java @@ -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(); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/GrpcClientChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/GrpcClientChannel.java index d62c5f80b0..dfc8faa859 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/GrpcClientChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/GrpcClientChannel.java @@ -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> 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} + *

+ * Case {@link RequestCode#CHECK_TRANSACTION_STATE} + * @see org.apache.rocketmq.broker.client.net.Broker2Client#checkProducerTransactionState + */ @Override public ChannelFuture writeAndFlush(Object msg) { CompletableFuture 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); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/common/Converter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/common/Converter.java index 8ffffcc597..ed103d1bd4 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/common/Converter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/common/Converter.java @@ -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 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; + } + } + } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java index c5bd4658d6..74b1349acc 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java @@ -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 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 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 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: