Golang: add push consumer (#990)

* push consumer

* fix test

* fix golang demo

* remove unused func in pushconsumer

* golang: optimization code

* golang: fix nil subscription in telemetry

* golang: update runtime verison 1.24

* golang: update unit test

* golang: fix ut

* golang: fix ut

* golang: fix build.yaml

---------

Co-authored-by: guyinyou <[email protected]>
Co-authored-by: rongtong <[email protected]>

Author: 3 people

.github/workflows/golang_build.yml

@@ -8,8 +8,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ ubuntu-22.04, windows-2022 ]
-        go: [1.17]
+        os: [ubuntu-22.04, windows-2022]
+        go: [1.24]
     steps:
       - name: Checkout
         uses: actions/checkout@v2

golang/client.go

@@ -31,11 +31,11 @@ import (
 	"github.com/apache/rocketmq-clients/golang/v5/pkg/ticker"
 	"github.com/apache/rocketmq-clients/golang/v5/pkg/utils"
 	v2 "github.com/apache/rocketmq-clients/golang/v5/protocol/v2"
-	"github.com/golang/protobuf/proto"
 	"github.com/google/uuid"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/proto"
 )
 
 type Client interface {
@@ -117,7 +117,7 @@ func (cs *defaultClientSession) startUp() {
 			if err != nil {
 				// we are recovering
 				if !cs.recovering {
-					cs.cli.log.Info("Encountered error while receiving TelemetryCommand, trying to recover")
+					cs.cli.log.Infof("Encountered error while receiving TelemetryCommand, trying to recover, err=%v", err)
 					// we wait five seconds to give time for the transmission error to be resolved externally before we attempt to read the message again.
 					time.Sleep(cs.recoveryWaitTime)
 					cs.recovering = true
@@ -176,13 +176,19 @@ func (cs *defaultClientSession) release() {
 }
 func (cs *defaultClientSession) publish(ctx context.Context, common *v2.TelemetryCommand) error {
 	var err error
-	cs.observerLock.RLock()
-	if cs.observer != nil {
-		err = cs.observer.Send(common)
-		cs.observerLock.RUnlock()
+
+	f0 := func() (bool, error) {
+		cs.observerLock.RLock()
+		defer cs.observerLock.RUnlock()
+		if cs.observer != nil {
+			return true, cs.observer.Send(common)
+		}
+		return false, nil
+	}
+	over, err := f0()
+	if over {
 		return err
 	}
-	cs.observerLock.RUnlock()
 
 	cs.observerLock.Lock()
 	defer cs.observerLock.Unlock()
@@ -218,8 +224,8 @@ type defaultClient struct {
 	endpointsTelemetryClientTable map[string]*defaultClientSession
 	endpointsTelemetryClientsLock sync.RWMutex
 	on                            atomic.Bool
-
-	clientImpl isClient
+	inited                        atomic.Bool
+	clientImpl                    isClient
 }
 
 var NewClient = func(config *Config, opts ...ClientOption) (Client, error) {
@@ -235,6 +241,7 @@ var NewClient = func(config *Config, opts ...ClientOption) (Client, error) {
 		messageInterceptors:           make([]MessageInterceptor, 0),
 		endpointsTelemetryClientTable: make(map[string]*defaultClientSession),
 		on:                            *atomic.NewBool(true),
+		inited:                        *atomic.NewBool(false),
 	}
 	cli.log = sugarBaseLogger.With("client_id", cli.clientID)
 	for _, opt := range opts {
@@ -299,7 +306,7 @@ func (cli *defaultClient) registerMessageInterceptor(messageInterceptor MessageI
 	cli.messageInterceptors = append(cli.messageInterceptors, messageInterceptor)
 }
 
-func (cli *defaultClient) doBefore(hookPoint MessageHookPoints, messageCommons []*MessageCommon) {
+func (cli *defaultClient) doBefore(hookPoint MessageHookPoints, messageCommons []*MessageCommon) error {
 	cli.messageInterceptorsLock.RLocker().Lock()
 	defer cli.messageInterceptorsLock.RLocker().Unlock()
 
@@ -309,9 +316,10 @@ func (cli *defaultClient) doBefore(hookPoint MessageHookPoints, messageCommons [
 			cli.log.Errorf("exception raised while intercepting message, hookPoint=%v, err=%v", hookPoint, err)
 		}
 	}
+	return nil
 }
 
-func (cli *defaultClient) doAfter(hookPoint MessageHookPoints, messageCommons []*MessageCommon, duration time.Duration, status MessageHookPointsStatus) {
+func (cli *defaultClient) doAfter(hookPoint MessageHookPoints, messageCommons []*MessageCommon, duration time.Duration, status MessageHookPointsStatus) error {
 	cli.messageInterceptorsLock.RLocker().Lock()
 	defer cli.messageInterceptorsLock.RLocker().Unlock()
 
@@ -321,6 +329,7 @@ func (cli *defaultClient) doAfter(hookPoint MessageHookPoints, messageCommons []
 			cli.log.Errorf("exception raised while intercepting message, hookPoint=%v, err=%v", hookPoint, err)
 		}
 	}
+	return nil
 }
 
 func (cli *defaultClient) getMessageQueues(ctx context.Context, topic string) ([]*v2.MessageQueue, error) {
@@ -417,6 +426,36 @@ func (cli *defaultClient) getSettingsCommand() *v2.TelemetryCommand {
 	}
 }
 
+func (cli *defaultClient) queryAssignments(ctx context.Context, topic string, group string, duration time.Duration) (*[]*v2.Assignment, error) {
+	ctx = cli.Sign(ctx)
+	response, err := cli.clientManager.QueryAssignments(ctx, cli.accessPoint, cli.getQueryAssignmentRequest(topic, group), duration)
+	if err != nil {
+		return nil, err
+	}
+	if response.GetStatus().GetCode() != v2.Code_OK {
+		return nil, &ErrRpcStatus{
+			Code:    int32(response.Status.GetCode()),
+			Message: response.GetStatus().GetMessage(),
+		}
+	}
+	ret := response.GetAssignments()
+	return &ret, nil
+}
+
+func (cli *defaultClient) getQueryAssignmentRequest(topic string, group string) *v2.QueryAssignmentRequest {
+	return &v2.QueryAssignmentRequest{
+		Topic: &v2.Resource{
+			Name:              topic,
+			ResourceNamespace: cli.config.NameSpace,
+		},
+		Group: &v2.Resource{
+			Name:              group,
+			ResourceNamespace: cli.config.NameSpace,
+		},
+		Endpoints: cli.accessPoint,
+	}
+}
+
 func (cli *defaultClient) doHeartbeat(target string, request *v2.HeartbeatRequest) error {
 	ctx := cli.Sign(context.Background())
 	endpoints, err := utils.ParseTarget(target)
@@ -428,9 +467,6 @@ func (cli *defaultClient) doHeartbeat(target string, request *v2.HeartbeatReques
 		return fmt.Errorf("failed to send heartbeat, endpoints=%v, err=%v, requestId=%s", endpoints, err, utils.GetRequestID(ctx))
 	}
 	if resp.Status.GetCode() != v2.Code_OK {
-		if resp.Status.GetCode() == v2.Code_UNRECOGNIZED_CLIENT_TYPE {
-			go cli.trySyncSettings()
-		}
 		cli.log.Errorf("failed to send heartbeat, code=%v, status message=[%s], endpoints=%v, requestId=%s", resp.Status.GetCode(), resp.Status.GetMessage(), endpoints, utils.GetRequestID(ctx))
 		return &ErrRpcStatus{
 			Code:    int32(resp.Status.GetCode()),
@@ -496,6 +532,7 @@ func (cli *defaultClient) startUp() error {
 	cm.startUp()
 	cm.RegisterClient(cli)
 	cli.clientManager = cm
+
 	for _, topic := range cli.initTopics {
 		_, err := cli.getMessageQueues(context.Background(), topic)
 		if err != nil {
@@ -546,6 +583,13 @@ func (cli *defaultClient) startUp() error {
 		})
 	}
 	ticker.Tick(f, time.Second*30, cli.done)
+
+	// wait syncSettings finish
+	for !cli.inited.Load() {
+		sugarBaseLogger.Infoln("wait for sync settings finish")
+		time.Sleep(time.Second)
+	}
+	sugarBaseLogger.Infoln("sync settings finished")
 	return nil
 }
 
@@ -587,6 +631,10 @@ func (cli *defaultClient) GracefulStop() error {
 	return nil
 }
 
+func (cli *defaultClient) isRunning() bool {
+	return cli.on.Load()
+}
+
 func (cli *defaultClient) Sign(ctx context.Context) context.Context {
 	now := time.Now().Format("20060102T150405Z")
 	return metadata.AppendToOutgoingContext(ctx,
@@ -629,7 +677,9 @@ func (cli *defaultClient) onSettingsCommand(endpoints *v2.Endpoints, settings *v
 	if metric != nil {
 		cli.clientMeterProvider.Reset(metric)
 	}
-	return cli.settings.applySettingsCommand(settings)
+	err := cli.settings.applySettingsCommand(settings)
+	cli.inited.Store(true)
+	return err
 }
 
 func (cli *defaultClient) onRecoverOrphanedTransactionCommand(endpoints *v2.Endpoints, command *v2.RecoverOrphanedTransactionCommand) {

golang/client_manager.go

@@ -33,6 +33,7 @@ type ClientManager interface {
 	RegisterClient(client Client)
 	UnRegisterClient(client Client)
 	QueryRoute(ctx context.Context, endpoints *v2.Endpoints, request *v2.QueryRouteRequest, duration time.Duration) (*v2.QueryRouteResponse, error)
+	QueryAssignments(ctx context.Context, endpoints *v2.Endpoints, request *v2.QueryAssignmentRequest, duration time.Duration) (*v2.QueryAssignmentResponse, error)
 	HeartBeat(ctx context.Context, endpoints *v2.Endpoints, request *v2.HeartbeatRequest, duration time.Duration) (*v2.HeartbeatResponse, error)
 	SendMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.SendMessageRequest, duration time.Duration) (*v2.SendMessageResponse, error)
 	Telemetry(ctx context.Context, endpoints *v2.Endpoints, duration time.Duration) (v2.MessagingService_TelemetryClient, error)
@@ -41,6 +42,7 @@ type ClientManager interface {
 	ReceiveMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.ReceiveMessageRequest) (v2.MessagingService_ReceiveMessageClient, error)
 	AckMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.AckMessageRequest, duration time.Duration) (*v2.AckMessageResponse, error)
 	ChangeInvisibleDuration(ctx context.Context, endpoints *v2.Endpoints, request *v2.ChangeInvisibleDurationRequest, duration time.Duration) (*v2.ChangeInvisibleDurationResponse, error)
+	ForwardMessageToDeadLetterQueue(ctx context.Context, endpoints *v2.Endpoints, request *v2.ForwardMessageToDeadLetterQueueRequest, duration time.Duration) (*v2.ForwardMessageToDeadLetterQueueResponse, error)
 }
 
 type clientManagerOptions struct {
@@ -232,6 +234,17 @@ func (cm *defaultClientManager) QueryRoute(ctx context.Context, endpoints *v2.En
 	return ret, err
 }
 
+func (cm *defaultClientManager) QueryAssignments(ctx context.Context, endpoints *v2.Endpoints, request *v2.QueryAssignmentRequest, duration time.Duration) (*v2.QueryAssignmentResponse, error) {
+	ctx, _ = context.WithTimeout(ctx, duration)
+	rpcClient, err := cm.getRpcClient(endpoints)
+	if err != nil {
+		return nil, err
+	}
+	ret, err := rpcClient.QueryAssignments(ctx, request)
+	cm.handleGrpcError(rpcClient, err)
+	return ret, err
+}
+
 func (cm *defaultClientManager) SendMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.SendMessageRequest, duration time.Duration) (*v2.SendMessageResponse, error) {
 	ctx, _ = context.WithTimeout(ctx, duration)
 	rpcClient, err := cm.getRpcClient(endpoints)
@@ -319,3 +332,14 @@ func (cm *defaultClientManager) ChangeInvisibleDuration(ctx context.Context, end
 	cm.handleGrpcError(rpcClient, err)
 	return ret, err
 }
+
+func (cm *defaultClientManager) ForwardMessageToDeadLetterQueue(ctx context.Context, endpoints *v2.Endpoints, request *v2.ForwardMessageToDeadLetterQueueRequest, duration time.Duration) (*v2.ForwardMessageToDeadLetterQueueResponse, error) {
+	ctx, _ = context.WithTimeout(ctx, duration)
+	rpcClient, err := cm.getRpcClient(endpoints)
+	if err != nil {
+		return nil, err
+	}
+	ret, err := rpcClient.ForwardMessageToDeadLetterQueue(ctx, request)
+	cm.handleGrpcError(rpcClient, err)
+	return ret, err
+}

golang/client_manager_mock.go

@@ -94,6 +94,21 @@ func (mr *MockClientManagerMockRecorder) EndTransaction(ctx, endpoints, request,
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EndTransaction", reflect.TypeOf((*MockClientManager)(nil).EndTransaction), ctx, endpoints, request, duration)
 }
 
+// ForwardMessageToDeadLetterQueue mocks base method.
+func (m *MockClientManager) ForwardMessageToDeadLetterQueue(ctx context.Context, endpoints *v2.Endpoints, request *v2.ForwardMessageToDeadLetterQueueRequest, duration time.Duration) (*v2.ForwardMessageToDeadLetterQueueResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ForwardMessageToDeadLetterQueue", ctx, endpoints, request, duration)
+	ret0, _ := ret[0].(*v2.ForwardMessageToDeadLetterQueueResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ForwardMessageToDeadLetterQueue indicates an expected call of ForwardMessageToDeadLetterQueue.
+func (mr *MockClientManagerMockRecorder) ForwardMessageToDeadLetterQueue(ctx, endpoints, request, duration interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ForwardMessageToDeadLetterQueue", reflect.TypeOf((*MockClientManager)(nil).ForwardMessageToDeadLetterQueue), ctx, endpoints, request, duration)
+}
+
 // HeartBeat mocks base method.
 func (m *MockClientManager) HeartBeat(ctx context.Context, endpoints *v2.Endpoints, request *v2.HeartbeatRequest, duration time.Duration) (*v2.HeartbeatResponse, error) {
 	m.ctrl.T.Helper()
@@ -124,6 +139,21 @@ func (mr *MockClientManagerMockRecorder) NotifyClientTermination(ctx, endpoints,
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NotifyClientTermination", reflect.TypeOf((*MockClientManager)(nil).NotifyClientTermination), ctx, endpoints, request, duration)
 }
 
+// QueryAssignments mocks base method.
+func (m *MockClientManager) QueryAssignments(ctx context.Context, endpoints *v2.Endpoints, request *v2.QueryAssignmentRequest, duration time.Duration) (*v2.QueryAssignmentResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "QueryAssignments", ctx, endpoints, request, duration)
+	ret0, _ := ret[0].(*v2.QueryAssignmentResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// QueryAssignments indicates an expected call of QueryAssignments.
+func (mr *MockClientManagerMockRecorder) QueryAssignments(ctx, endpoints, request, duration interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "QueryAssignments", reflect.TypeOf((*MockClientManager)(nil).QueryAssignments), ctx, endpoints, request, duration)
+}
+
 // QueryRoute mocks base method.
 func (m *MockClientManager) QueryRoute(ctx context.Context, endpoints *v2.Endpoints, request *v2.QueryRouteRequest, duration time.Duration) (*v2.QueryRouteResponse, error) {
 	m.ctrl.T.Helper()

golang/client_manager_test.go

@@ -338,7 +338,7 @@ func TestCMClearIdleRpcClients(t *testing.T) {
 	})
 	defer stubs.Reset()
 
-	MOCK_RPC_CLIENT.EXPECT().idleDuration().Return(time.Hour * 24 * 365)
+	MOCK_RPC_CLIENT.EXPECT().idleDuration().Return(time.Hour * 24 * 365).AnyTimes()
 	cm := NewDefaultClientManager()
 	cm.startUp()
 	cm.RegisterClient(MOCK_CLIENT)

golang/client_test.go

@@ -74,6 +74,7 @@ func BuildCLient(t *testing.T) *defaultClient {
 	if err != nil {
 		t.Error(err)
 	}
+	cli.inited.Store(true)
 	err = cli.startUp()
 	if err != nil {
 		t.Error(err)
@@ -140,6 +141,7 @@ func TestCLINewClient(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
+	cli.(*defaultClient).inited.Store(true)
 	sugarBaseLogger.Info(cli)
 	err = cli.(*defaultClient).startUp()
 	if err != nil {
@@ -223,7 +225,6 @@ func TestRestoreDefaultClientSessionZeroErrors(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	default_cli_session.publish(context.TODO(), &v2.TelemetryCommand{})
 	observedLogs := PrepareTestLogger(cli)
 	default_cli_session.observer = &MOCK_MessagingService_TelemetryClient{
 		recv_error_count: 0,
@@ -232,6 +233,8 @@ func TestRestoreDefaultClientSessionZeroErrors(t *testing.T) {
 	default_cli_session.recoveryWaitTime = time.Second
 	cli.settings = &simpleConsumerSettings{}
 
+	default_cli_session.publish(context.TODO(), &v2.TelemetryCommand{})
+
 	// when
 	time.Sleep(3 * time.Second)
 
@@ -249,7 +252,6 @@ func TestRestoreDefaultClientSessionOneError(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	default_cli_session.publish(context.TODO(), &v2.TelemetryCommand{})
 	observedLogs := PrepareTestLogger(cli)
 	default_cli_session.observer = &MOCK_MessagingService_TelemetryClient{
 		recv_error_count: 1,
@@ -258,13 +260,14 @@ func TestRestoreDefaultClientSessionOneError(t *testing.T) {
 	default_cli_session.recoveryWaitTime = time.Second
 	cli.settings = &simpleConsumerSettings{}
 
+	default_cli_session.publish(context.TODO(), &v2.TelemetryCommand{})
 	// when
-	time.Sleep(3 * time.Second)
+	time.Sleep(4 * time.Second)
 
 	// then
 	sugarBaseLogger.Info(observedLogs.All())
 	commandExecutionLog := observedLogs.All()[:3]
-	assert.Equal(t, "Encountered error while receiving TelemetryCommand, trying to recover", commandExecutionLog[0].Message)
+	assert.Equal(t, "Encountered error while receiving TelemetryCommand, trying to recover, err=EOF", commandExecutionLog[0].Message)
 	assert.Equal(t, "Managed to recover, executing message", commandExecutionLog[1].Message)
 	assert.Equal(t, "Executed command successfully", commandExecutionLog[2].Message)
 }
@@ -276,7 +279,6 @@ func TestRestoreDefaultClientSessionTwoErrors(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	default_cli_session.publish(context.TODO(), &v2.TelemetryCommand{})
 	observedLogs := PrepareTestLogger(cli)
 	default_cli_session.observer = &MOCK_MessagingService_TelemetryClient{
 		recv_error_count: 2,
@@ -285,13 +287,14 @@ func TestRestoreDefaultClientSessionTwoErrors(t *testing.T) {
 	default_cli_session.recoveryWaitTime = time.Second
 	cli.settings = &simpleConsumerSettings{}
 
+	default_cli_session.publish(context.TODO(), &v2.TelemetryCommand{})
 	// when
-	time.Sleep(3 * time.Second)
+	time.Sleep(4 * time.Second)
 
 	// then
 	sugarBaseLogger.Info(observedLogs.All())
 	commandExecutionLog := observedLogs.All()[:2]
-	assert.Equal(t, "Encountered error while receiving TelemetryCommand, trying to recover", commandExecutionLog[0].Message)
+	assert.Equal(t, "Encountered error while receiving TelemetryCommand, trying to recover, err=EOF", commandExecutionLog[0].Message)
 	assert.Equal(t, "Failed to recover, err=EOF", commandExecutionLog[1].Message)
 }
 

golang/consumer.go

@@ -26,4 +26,5 @@ import (
 type Consumer interface {
 	GetGroupName() string
 	wrapReceiveMessageRequest(batchSize int, messageQueue *v2.MessageQueue, filterExpression *FilterExpression, invisibleDuration time.Duration) *v2.ReceiveMessageRequest
+	isClient
 }