thunguo commented on code in PR #1054: URL: https://github.com/apache/incubator-seata-go/pull/1054#discussion_r2872473019
########## pkg/integration/rocketmq/seata_producer.go: ########## @@ -0,0 +1,157 @@ +/* + * 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 rocketmq + +import ( + "context" + "fmt" + "sync" + + "github.com/apache/rocketmq-client-go/v2/primitive" + "github.com/apache/rocketmq-client-go/v2/producer" + + "seata.apache.org/seata-go/v2/pkg/rm/tcc" + "seata.apache.org/seata-go/v2/pkg/tm" + "seata.apache.org/seata-go/v2/pkg/util/log" +) + +type transactionProducerInterface interface { + Start() error + Shutdown() error + SendMessageInTransaction(context.Context, *primitive.Message) (*primitive.TransactionSendResult, error) +} + +type SeataMQProducer struct { + config *SeataMQProducerConfig + transactionProducer transactionProducerInterface + tccAction *TCCRocketMQAction + tccProxy *tcc.TCCServiceProxy + + mu sync.RWMutex + closed bool +} + +func NewSeataMQProducer(cfg *SeataMQProducerConfig) (*SeataMQProducer, error) { + if cfg == nil { + return nil, fmt.Errorf("config cannot be nil") + } + + if cfg.NameServerAddrs == nil || len(cfg.NameServerAddrs) == 0 { + return nil, fmt.Errorf("NameServerAddrs cannot be empty") + } + + if cfg.GroupName == "" { + return nil, fmt.Errorf("GroupName cannot be empty") + } + + p := &SeataMQProducer{ + config: cfg, + } + + p.tccAction = NewTCCRocketMQAction(p) + + tccResource, err := tcc.ParseTCCResource(p.tccAction) Review Comment: Is `ParseTCCResource` being called repeatedly here? `tcc.NewTCCServiceProxy` calls `ParseTCCResource` again. ########## pkg/integration/rocketmq/transaction_listener.go: ########## @@ -0,0 +1,86 @@ +/* + * 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 rocketmq + +import ( + "github.com/apache/rocketmq-client-go/v2/primitive" + + "seata.apache.org/seata-go/v2/pkg/constant" + "seata.apache.org/seata-go/v2/pkg/protocol/message" + "seata.apache.org/seata-go/v2/pkg/remoting/getty" + "seata.apache.org/seata-go/v2/pkg/util/log" +) + +type SeataTransactionListener struct { + producer *SeataMQProducer +} + +func NewSeataTransactionListener(producer *SeataMQProducer) *SeataTransactionListener { + return &SeataTransactionListener{producer: producer} +} + +func (l *SeataTransactionListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState { + xid := msg.GetProperty(constant.PropertySeataXID) + if xid == "" { + return primitive.CommitMessageState + } + log.Debugf("[SeataTransactionListener] ExecuteLocalTransaction, xid=%s, returning UnknownState", xid) + return primitive.UnknowState +} + +func (l *SeataTransactionListener) CheckLocalTransaction(msgExt *primitive.MessageExt) primitive.LocalTransactionState { + xid := msgExt.GetProperty(constant.PropertySeataXID) + if xid == "" { + log.Warnf("[SeataTransactionListener] CheckLocalTransaction: missing XID, rollback") + return primitive.RollbackMessageState + } + + branchIdStr := msgExt.GetProperty(constant.PropertySeataBranchId) + log.Infof("[SeataTransactionListener] CheckLocalTransaction, xid=%s, branchId=%s", xid, branchIdStr) + + globalStatus, err := l.queryGlobalStatus(xid) + if err != nil { + log.Errorf("[SeataTransactionListener] Query global status failed, xid=%s, err=%v", xid, err) + return primitive.UnknowState + } + + switch globalStatus { + case message.GlobalStatusCommitted: + log.Infof("[SeataTransactionListener] Global tx committed, xid=%s", xid) + return primitive.CommitMessageState + case message.GlobalStatusRollbacked, message.GlobalStatusTimeoutRollbacked, message.GlobalStatusRollbackFailed: + log.Infof("[SeataTransactionListener] Global tx rollbacked, xid=%s, status=%v", xid, globalStatus) + return primitive.RollbackMessageState + default: Review Comment: Do you need to handle intermediate states like GlobalStatusBegin, Committing, and Rollbacking? ########## pkg/integration/rocketmq/producer_factory.go: ########## @@ -0,0 +1,82 @@ +/* + * 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 rocketmq + +import ( + "fmt" + "sync" +) + +var ( + globalProducer *SeataMQProducer + producerMutex sync.RWMutex + producerInitOnce sync.Once + producerInitialized bool +) + +func InitSeataMQProducer(cfg *SeataMQProducerConfig) error { + var initErr error + producerInitOnce.Do(func() { Review Comment: Wouldn't it be better to manually control `Init` here? If `Once` fails, it won't be executed again, instead, it will simply be set to `nil`, and the error will be delayed until the actual action occurs. ########## pkg/integration/rocketmq/transaction_listener.go: ########## @@ -0,0 +1,86 @@ +/* + * 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 rocketmq + +import ( + "github.com/apache/rocketmq-client-go/v2/primitive" + + "seata.apache.org/seata-go/v2/pkg/constant" + "seata.apache.org/seata-go/v2/pkg/protocol/message" + "seata.apache.org/seata-go/v2/pkg/remoting/getty" + "seata.apache.org/seata-go/v2/pkg/util/log" +) + +type SeataTransactionListener struct { + producer *SeataMQProducer +} + +func NewSeataTransactionListener(producer *SeataMQProducer) *SeataTransactionListener { + return &SeataTransactionListener{producer: producer} +} + +func (l *SeataTransactionListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState { + xid := msg.GetProperty(constant.PropertySeataXID) + if xid == "" { + return primitive.CommitMessageState + } + log.Debugf("[SeataTransactionListener] ExecuteLocalTransaction, xid=%s, returning UnknownState", xid) + return primitive.UnknowState +} + +func (l *SeataTransactionListener) CheckLocalTransaction(msgExt *primitive.MessageExt) primitive.LocalTransactionState { + xid := msgExt.GetProperty(constant.PropertySeataXID) + if xid == "" { + log.Warnf("[SeataTransactionListener] CheckLocalTransaction: missing XID, rollback") + return primitive.RollbackMessageState + } + + branchIdStr := msgExt.GetProperty(constant.PropertySeataBranchId) + log.Infof("[SeataTransactionListener] CheckLocalTransaction, xid=%s, branchId=%s", xid, branchIdStr) + + globalStatus, err := l.queryGlobalStatus(xid) + if err != nil { + log.Errorf("[SeataTransactionListener] Query global status failed, xid=%s, err=%v", xid, err) + return primitive.UnknowState + } + + switch globalStatus { + case message.GlobalStatusCommitted: + log.Infof("[SeataTransactionListener] Global tx committed, xid=%s", xid) + return primitive.CommitMessageState + case message.GlobalStatusRollbacked, message.GlobalStatusTimeoutRollbacked, message.GlobalStatusRollbackFailed: + log.Infof("[SeataTransactionListener] Global tx rollbacked, xid=%s, status=%v", xid, globalStatus) + return primitive.RollbackMessageState + default: + log.Infof("[SeataTransactionListener] Global tx in progress, xid=%s, status=%v", xid, globalStatus) + return primitive.UnknowState + } +} + +func (l *SeataTransactionListener) queryGlobalStatus(xid string) (message.GlobalStatus, error) { + req := message.GlobalStatusRequest{ + AbstractGlobalEndRequest: message.AbstractGlobalEndRequest{ + Xid: xid, + }, + } + res, err := getty.GetGettyRemotingClient().SendSyncRequest(req) + if err != nil { + return message.GlobalStatusUnKnown, err + } + return res.(message.GlobalStatusResponse).GlobalStatus, nil Review Comment: Avoid using the approach `res.(message.GlobalStatusResponse).GlobalStatus`. First check the assertion, then retrieve the value to prevent panics that cannot be handled on the main chain, as follows: ```go gsResp, ok := res.(message.GlobalStatusResponse) if !ok { // ... } return gsResp.GlobalStatus, nil ``` ########## pkg/integration/rocketmq/tcc_rocketmq_action.go: ########## @@ -0,0 +1,92 @@ +/* + * 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 rocketmq + +import ( + "context" + "fmt" + + "github.com/apache/rocketmq-client-go/v2/primitive" + + "seata.apache.org/seata-go/v2/pkg/constant" + "seata.apache.org/seata-go/v2/pkg/tm" + "seata.apache.org/seata-go/v2/pkg/util/log" +) + +type TCCRocketMQAction struct { + producer *SeataMQProducer +} + +func NewTCCRocketMQAction(producer *SeataMQProducer) *TCCRocketMQAction { + return &TCCRocketMQAction{ + producer: producer, + } +} + +func (a *TCCRocketMQAction) GetActionName() string { + return ResourceIDTCCRocketMQ +} + +func (a *TCCRocketMQAction) Prepare(ctx context.Context, params interface{}) (bool, error) { + msg, ok := params.(*primitive.Message) + if !ok { + return false, fmt.Errorf("params must be *primitive.Message, got %T", params) + } + + bac := tm.GetBusinessActionContext(ctx) + if bac == nil { + return false, fmt.Errorf("BusinessActionContext not found in context") + } + + xid := tm.GetXID(ctx) + if xid == "" { + return false, fmt.Errorf("XID not found in context") + } + + msg.WithProperty(constant.PropertySeataXID, xid) + msg.WithProperty(constant.PropertySeataBranchId, fmt.Sprintf("%d", bac.BranchId)) + + result, err := a.producer.transactionProducer.SendMessageInTransaction(ctx, msg) + if err != nil { + log.Errorf("[TCCRocketMQ] Prepare failed, xid=%s, err=%v", xid, err) + return false, err + } + + bac.ActionContext[ActionContextKeyMsgId] = result.MsgID + bac.ActionContext[ActionContextKeyOffsetMsgId] = result.OffsetMsgID + bac.ActionContext[ActionContextKeyQueueOffset] = result.QueueOffset + bac.ActionContext[ActionContextKeyTransactionId] = result.TransactionID + if result.MessageQueue != nil { + bac.ActionContext[ActionContextKeyQueueId] = result.MessageQueue.QueueId + bac.ActionContext[ActionContextKeyBrokerName] = result.MessageQueue.BrokerName + } + + log.Infof("[TCCRocketMQ] Prepare success, xid=%s, branchId=%d, msgId=%s", xid, bac.BranchId, result.MsgID) + + return true, nil +} + +func (a *TCCRocketMQAction) Commit(ctx context.Context, bac *tm.BusinessActionContext) (bool, error) { Review Comment: Why do both `Commit` and `Rollback` return `true` here? Doesn't this seem to contradict the semantics of TCC? ########## pkg/integration/rocketmq/seata_producer.go: ########## @@ -0,0 +1,157 @@ +/* + * 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 rocketmq + +import ( + "context" + "fmt" + "sync" + + "github.com/apache/rocketmq-client-go/v2/primitive" + "github.com/apache/rocketmq-client-go/v2/producer" + + "seata.apache.org/seata-go/v2/pkg/rm/tcc" + "seata.apache.org/seata-go/v2/pkg/tm" + "seata.apache.org/seata-go/v2/pkg/util/log" +) + +type transactionProducerInterface interface { + Start() error + Shutdown() error + SendMessageInTransaction(context.Context, *primitive.Message) (*primitive.TransactionSendResult, error) +} + +type SeataMQProducer struct { + config *SeataMQProducerConfig + transactionProducer transactionProducerInterface + tccAction *TCCRocketMQAction + tccProxy *tcc.TCCServiceProxy + + mu sync.RWMutex + closed bool +} + +func NewSeataMQProducer(cfg *SeataMQProducerConfig) (*SeataMQProducer, error) { + if cfg == nil { + return nil, fmt.Errorf("config cannot be nil") + } + + if cfg.NameServerAddrs == nil || len(cfg.NameServerAddrs) == 0 { + return nil, fmt.Errorf("NameServerAddrs cannot be empty") + } + + if cfg.GroupName == "" { + return nil, fmt.Errorf("GroupName cannot be empty") + } + + p := &SeataMQProducer{ + config: cfg, + } + + p.tccAction = NewTCCRocketMQAction(p) + + tccResource, err := tcc.ParseTCCResource(p.tccAction) + if err != nil { + return nil, fmt.Errorf("parse TCC resource failed: %w", err) + } + + p.tccProxy, err = tcc.NewTCCServiceProxy(tccResource) + if err != nil { + return nil, fmt.Errorf("create TCC proxy failed: %w", err) + } + + listener := NewSeataTransactionListener(p) + opts := cfg.ToRocketMQProducerOptions() + + p.transactionProducer, err = producer.NewTransactionProducer(listener, opts...) + if err != nil { + return nil, fmt.Errorf("create transaction producer failed: %w", err) + } + + return p, nil +} + +func (p *SeataMQProducer) Start() error { + p.mu.Lock() + defer p.mu.Unlock() + + if p.closed { + return fmt.Errorf("producer already closed") + } + + return p.transactionProducer.Start() +} + +func (p *SeataMQProducer) Shutdown() error { + p.mu.Lock() + defer p.mu.Unlock() + + if p.closed { + return nil + } + + p.closed = true + return p.transactionProducer.Shutdown() +} + +func (p *SeataMQProducer) Send(ctx context.Context, msg *primitive.Message) (*primitive.SendResult, error) { + p.mu.RLock() + defer p.mu.RUnlock() + + if p.closed { + return nil, fmt.Errorf("producer is closed") + } + + if msg == nil { + return nil, fmt.Errorf("message cannot be nil") + } + + if !tm.IsGlobalTx(ctx) { + return p.sendSync(ctx, msg) + } + + _, err := p.tccProxy.Prepare(ctx, msg) + if err != nil { + log.Errorf("[SeataMQProducer] Send in global tx failed, xid=%s, err=%v", tm.GetXID(ctx), err) + return nil, err + } + + bac := tm.GetBusinessActionContext(ctx) + return &primitive.SendResult{ + Status: primitive.SendOK, + MsgID: getStringFromMap(bac.ActionContext, ActionContextKeyMsgId), + OffsetMsgID: getStringFromMap(bac.ActionContext, ActionContextKeyOffsetMsgId), + }, nil +} + +func (p *SeataMQProducer) sendSync(ctx context.Context, msg *primitive.Message) (*primitive.SendResult, error) { Review Comment: There seems to be a semantic issue with `sendSync` here, as it internally uses `SendMessageInTransaction`. This causes the message to pass through `ExecuteLocalTransaction`, at which point `xid` is empty, triggering a commit. Currently, this process seems fine, but if the logic of `ExecuteLocalTransaction` is modified later, this behavior might become abnormal. You could embed a `rocketmq.NewProducer` within `SeataMQProducer` to handle common messages. ########## pkg/integration/rocketmq/constants.go: ########## @@ -0,0 +1,32 @@ +/* + * 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 rocketmq + +const ( + ResourceIDTCCRocketMQ = "tccRocketMQ" + + ActionContextKeyMessage = "message" + ActionContextKeySendResult = "sendResult" Review Comment: It appears that the constant `ActionContextKeySendResult` has not been used. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
