Copilot commented on code in PR #1054:
URL:
https://github.com/apache/incubator-seata-go/pull/1054#discussion_r2986830795
##########
pkg/remoting/getty/getty_remoting.go:
##########
@@ -128,7 +128,11 @@ func (g *GettyRemoting)
NotifyRpcMessageResponse(rpcMessage message.RpcMessage)
messageFuture.Response = rpcMessage.Body
// todo add messageFuture.Err
// messageFuture.Err = rpcMessage.Err
- messageFuture.Done <- struct{}{}
+ select {
+ case messageFuture.Done <- struct{}{}:
+ default:
+ log.Warnf("response arrived after timeout for msg ID:
%d", rpcMessage.ID)
Review Comment:
The `default` branch here triggers when `Done` already has a signal (buffer
full), which is not necessarily “response arrived after timeout”. This can make
the warning misleading (e.g., duplicate/late responses). Consider adjusting the
log message to reflect the actual condition (e.g., “future already completed;
dropping signal”) or tracking timeout state explicitly if you want to log
timeouts.
```suggestion
log.Warnf("message future already completed; dropping
additional response for msg ID: %d", rpcMessage.ID)
```
##########
pkg/integration/rocketmq/transaction_listener.go:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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"
+
+ "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.GlobalStatusCommitting, message.GlobalStatusBegin:
+ log.Infof("[SeataTransactionListener] Global tx committing/in
progress, xid=%s, status=%v", xid, globalStatus)
+ return primitive.UnknowState
+ case message.GlobalStatusRollbacking:
+ log.Infof("[SeataTransactionListener] Global tx rolling back,
xid=%s, status=%v", xid, globalStatus)
+ return primitive.UnknowState
+ case message.GlobalStatusRollbacked,
message.GlobalStatusTimeoutRollbacked, message.GlobalStatusRollbackFailed:
+ log.Infof("[SeataTransactionListener] Global tx rollbacked,
xid=%s, status=%v", xid, globalStatus)
Review Comment:
`CheckLocalTransaction` only handles a subset of `message.GlobalStatus`
values. Final statuses like `GlobalStatusCommitFailed`,
`GlobalStatusTimeoutRollbackFailed`, and `GlobalStatusFinished`, and
transitional ones like `GlobalStatusAsyncCommitting` /
`GlobalStatusCommitRetrying` / `GlobalStatusRollbackRetrying` currently fall
into `default` and return `UnknowState`, which can cause RocketMQ to keep
checking indefinitely. Please map the remaining statuses explicitly to
Commit/Rollback/Unknown according to Seata semantics (and ideally document the
mapping).
```suggestion
// Map Seata global transaction status to RocketMQ local transaction
state.
// - Committed / Finished -> CommitMessageState
// - Rollbacked / rollback failed -> RollbackMessageState
// - Begin / *Committing / *Rollbacking (including retry/async) ->
UnknowState (broker will re-check)
switch globalStatus {
case message.GlobalStatusCommitted, message.GlobalStatusFinished:
log.Infof("[SeataTransactionListener] Global tx
committed/finished, xid=%s, status=%v", xid, globalStatus)
return primitive.CommitMessageState
case message.GlobalStatusCommitting,
message.GlobalStatusBegin,
message.GlobalStatusAsyncCommitting,
message.GlobalStatusCommitRetrying:
log.Infof("[SeataTransactionListener] Global tx committing/in
progress, xid=%s, status=%v", xid, globalStatus)
return primitive.UnknowState
case message.GlobalStatusRollbacking,
message.GlobalStatusRollbackRetrying,
message.GlobalStatusTimeoutRollbacking,
message.GlobalStatusTimeoutRollbackRetrying:
log.Infof("[SeataTransactionListener] Global tx rolling back/in
progress, xid=%s, status=%v", xid, globalStatus)
return primitive.UnknowState
case message.GlobalStatusRollbacked,
message.GlobalStatusTimeoutRollbacked,
message.GlobalStatusRollbackFailed,
message.GlobalStatusCommitFailed,
message.GlobalStatusTimeoutRollbackFailed:
log.Infof("[SeataTransactionListener] Global tx
rollbacked/failed, xid=%s, status=%v", xid, globalStatus)
```
##########
go.mod:
##########
@@ -52,6 +52,7 @@ require (
github.com/Workiva/go-datastructures v1.0.52 // indirect
github.com/antlr/antlr4/runtime/Go/antlr/v4
v4.0.0-20230305170008-8188dc5388df // indirect
github.com/apache/dubbo-go-hessian2 v1.11.4 // indirect
+ github.com/apache/rocketmq-client-go/v2 v2.1.2 // indirect
Review Comment:
`github.com/apache/rocketmq-client-go/v2` is imported directly by the new
RocketMQ integration, so it should not be marked as `// indirect` in go.mod.
Please run `go mod tidy` (or update the require line) so direct dependencies
are recorded correctly.
```suggestion
github.com/apache/rocketmq-client-go/v2 v2.1.2
```
##########
pkg/integration/rocketmq/seata_producer.go:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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 normalProducerInterface interface {
+ Start() error
+ Shutdown() error
+ SendSync(ctx context.Context, msg ...*primitive.Message)
(*primitive.SendResult, error)
+}
+
+type SeataMQProducer struct {
+ config *SeataMQProducerConfig
+ transactionProducer transactionProducerInterface
+ normalProducer normalProducerInterface
+ 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)
+
+ // NewTCCServiceProxy internally calls ParseTCCResource, so we pass the
action directly
+ var err error
+ p.tccProxy, err = tcc.NewTCCServiceProxy(p.tccAction)
+ 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)
+ }
+
+ normalCfg := *cfg
+ normalCfg.GroupName = cfg.GroupName + "-normal"
+ normalOpts := normalCfg.ToRocketMQProducerOptions()
+ p.normalProducer, err = producer.NewDefaultProducer(normalOpts...)
+ if err != nil {
+ return nil, fmt.Errorf("create normal 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")
+ }
+
+ if err := p.transactionProducer.Start(); err != nil {
+ return err
+ }
+
+ if err := p.normalProducer.Start(); err != nil {
+ p.transactionProducer.Shutdown()
+ return err
+ }
+
+ return nil
+}
+
+func (p *SeataMQProducer) Shutdown() error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.closed {
+ return nil
+ }
+
+ p.closed = true
+
+ var errs []error
+ if err := p.transactionProducer.Shutdown(); err != nil {
+ errs = append(errs, err)
+ }
+ if err := p.normalProducer.Shutdown(); err != nil {
+ errs = append(errs, err)
+ }
+
+ if len(errs) > 0 {
+ return fmt.Errorf("shutdown errors: %v", errs)
+ }
+ return nil
+}
+
+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
+}
Review Comment:
This new global-tx send path (`tm.IsGlobalTx` → `tccProxy.Prepare` →
transactional half-message) and the returned `SendResult` are not covered by
tests. Adding unit tests with stubbed `transactionProducer` / `normalProducer`
(and a mocked Getty client for status queries) would help validate property
injection (XID/BranchId), error paths, and the behavior difference between
global vs non-global contexts.
--
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]
