apache/rocketmq-clients

RocketMQ Clients - Collection of Client Bindings for Apache RocketMQ

RocketMQ Clients - Collection of Client Bindings for Apache RocketMQ

English | 简体中文 | RocketMQ Website

Overview

Client bindings for Apache RocketMQ, as known as RocketMQ 5.x SDK. All of them follow the specification of rocketmq-apis, replacing 4.x remoting-based counterparts. Clients in this repository are built on top of Protocol Buffers and gRPC.

Goal

Provide cloud-native and robust solutions for Java, C++, C#, Golang, Rust and all other mainstream programming languages.

Features and Status

• Ready - ✅
• Work in progress - 🚧

Feature	Java	C/C++	C#	Golang	Rust	Python	Node.js	PHP
Producer with standard messages	✅	✅	✅	✅	✅	🚧	✅	🚧
Producer with FIFO messages	✅	✅	✅	✅	✅	🚧	✅	🚧
Producer with timed/delay messages	✅	✅	✅	✅	✅	🚧	✅	🚧
Producer with transactional messages	✅	✅	✅	✅	✅	🚧	✅	🚧
Simple consumer	✅	✅	✅	✅	✅	🚧	✅	🚧
Push consumer with concurrent message listener	✅	✅	🚧	🚧	🚧	🚧	🚧	🚧
Push consumer with FIFO message listener	✅	✅	🚧	🚧	🚧	🚧	🚧	🚧

Prerequisite and Build

This project is organized as a monorepo, with each language's bindings containing specific instructions in their respective subdirectories for building the project. As the rocketmq-apis submodule is incorporated into the project and could be utilized by some bindings, we highly advise cloning the repository by executing the command below:

git clone --recursive [email protected]:apache/rocketmq-clients.git

Contributing

Similar to other projects of Apache RocketMQ, any attempt to make this project better is welcome, including but not limited to filing a bug report, correcting type error or document writing to complete feature implementation. Do not hesitate to make a pull request if this project catches your attention.

Related

• rocketmq: The implementation of server-side.
• rocketmq-apis: Common communication protocol between server and client.
• RIP-37: New and Unified APIs: RocketMQ proposal of new and unified APIs crossing different languages.
• RIP-39: Support gRPC protocol: RocketMQ proposal of gRPC protocol support.

License

Apache License, Version 2.0 Copyright (C) Apache Software Foundation

rocketmq-clients's Issues

Add docs about the design of message identifier.

NPE occurs when the instance username and password are not set

I use 5.0 sdk but refer to how 4.0 is used (like here: https://help.aliyun.com/document_detail/446709.html)

When my instanceUsername and password are both null and I access the Alibaba Cloud MQ instance from vpc, there is an NPE.

golang DATA RACE

// github.com/apache/rocketmq-clients/[email protected]/rpc_client.go:98
// github.com/apache/rocketmq-clients/[email protected]/rpc_client.go:98
func (rc *rpcClient) QueryRoute(ctx context.Context, request *v2.QueryRouteRequest) (*v2.QueryRouteResponse, error) {
	rc.activityNanoTime = time.Now() // DATA RACE
	resp, err := rc.msc.QueryRoute(ctx, request)
	sugarBaseLogger.Debugf("queryRoute request: %v, response: %v, err: %v", request, resp, err)
	return resp, err
}

==================
==================
// github.com/apache/rocketmq-clients/[email protected]/rpc_client.go:137
// github.com/apache/rocketmq-clients/[email protected]/rpc_client.go:137
func (rc *rpcClient) ReceiveMessage(ctx context.Context, request *v2.ReceiveMessageRequest) (v2.MessagingService_ReceiveMessageClient, error) {
	rc.activityNanoTime = time.Now() // DATA RACE
	resp, err := rc.msc.ReceiveMessage(ctx, request)
	sugarBaseLogger.Debugf("receiveMessage request: %v, err: %v", request, err)
	return resp, err
}

==================
==================

// github.com/apache/rocketmq-clients/[email protected]/metric.go:166
func (dcmp *defaultClientMeterProvider) isEnabled() bool {
	return dcmp.clientMeter.enabled // DATA RACE
}

// github.com/apache/rocketmq-clients/[email protected]/metric.go:183
	if !metric.GetOn() {
		dcmp.clientMeter.shutdown()
		sugarBaseLogger.Infof("metric is off, clientId=%s", dcmp.client.GetClientID())
		dcmp.clientMeter = NewDefaultClientMeter(nil, false, nil, dcmp.client.GetClientID()) // DATA RACE
		return
	}

==================
==================

// github.com/apache/rocketmq-clients/[email protected]/producer_options.go:117
func (ps *producerSettings) IsValidateMessageType() bool {
	return ps.validateMessageType // DATA RACE
}

// github.com/apache/rocketmq-clients/[email protected]/producer_options.go:163
	}
	ps.validateMessageType = v.Publishing.GetValidateMessageType()
	ps.maxBodySizeBytes = int(v.Publishing.GetMaxBodySize()) // DATA RACE

==================
==================

// github.com/apache/rocketmq-clients/[email protected]/message_id_codec.go:138
// github.com/apache/rocketmq-clients/[email protected]/message_id_codec.go:139
	deltaSeconds := deltaSeconds()
	if seconds != deltaSeconds { // DATA RACE
		seconds = deltaSeconds // DATA RACE
	}

==================
==================

// github.com/apache/rocketmq-clients/[email protected]/producer.go:288 
		if uMsg.pubMsg == nil {
			pubMessage, err = NewPublishingMessage(msg, p.pSetting, txEnabled)  // DATA RACE
			if err != nil {
				return nil, err
			}
		}
// github.com/apache/rocketmq-clients/[email protected]/producer_options.go:164
	ps.validateMessageType = v.Publishing.GetValidateMessageType()
	ps.maxBodySizeBytes = int(v.Publishing.GetMaxBodySize())  // DATA RACE

Doc: add client id detailed design

Golang: onVerifyMessageCommand

When the server sends a consumption verification request, it will go to the onVerifyMessageCommand function (only the pushconsumer needs to complete the real consumption operation), which theoretically will not be sent to the producer, but it is recommended to implement a fast return in the producer to reduce the wait on the server.

Add a unified doc set to introduce the design and implementation of clients

To help all the clients of different languages to follow the same design and limitations, we should have a doc set to describe this information, including:

1. The design philosophy, e.g., immutable API, lightweight, and so on
2. The model introduction
3. The message type and the detailed process design
4. The consumer kind, e.g., PushConsumer, SimpleConsumer, and their advanced feature set, like order, transactional
5. Error handling, retry, exception, error, and so on
6. Core message flow, producer & consumer flow

Golang: GenMessageId

The messageid generated when sending a message is currently uuid, while the java sdk is generated in MessageIdCodec.getInstance().nextMessageId(). It is recommended to refer to the alignment.

Could MessageView provide a graceful method to get message body

now messageView.getBody() returns a ByteBuffer object and can't directly get body byte array from ByteBuffer

Java: Make it possible to control client shutdown in JVM shutdown hook

ClientImpl‘s Constructor

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            LOGGER.info("JVM shutdown hook is invoked, clientId={}, state={}", clientId, ClientImpl.this.state());
            ClientImpl.this.stopAsync().awaitTerminated();
        }));

Whether this hook can be managed by the user？

Is it possible to delete or add a parameter to control？

Doc: add Ack/ChangeInvisibleTime retry policy for PushConsumer

Fix the coverage badge

It seems that the coverage badge of java/cpp does not working.

Client log file location specification

It's necessary to have a specification about client log file location for all kinds of clients and different languages.

Do we need a wiki or README?

Cannot check peer: missing selected ALPN property

Following installation guide of rocketmq-5.0 and rocketmq-client/cpp 5.0, I got following error while running example of CPP client.

[mikehuang@fedora build]$ ./examples/example_push_consumer --access_point=127.0.0.1:9876 --topic=T_TestTopic  --access_key=RocketMQ2 --access_secret=12345678
RocketMQ log files path: /home/mikehuang/logs/rocketmq
[2022/11/14-10:25:45.083 +08:00] [rocketmq_logger] [---I---] [thread 50329] ClientManager[ResourceNamespace=] created /home/mikehuang/Downloads/rocketmq-clients/cpp/source/client/ClientManagerImpl.cpp:86
[2022/11/14-10:25:45.084 +08:00] [rocketmq_logger] [---I---] [thread 50329] Scheduler threads start to loop /home/mikehuang/Downloads/rocketmq-clients/cpp/source/scheduler/SchedulerImpl.cpp:92
[2022/11/14-10:25:45.086 +08:00] [rocketmq_logger] [---I---] [thread 50329] Create a RPC client to ipv4:127.0.0.1:9876 /home/mikehuang/Downloads/rocketmq-clients/cpp/source/client/ClientManagerImpl.cpp:483
[2022/11/14-10:25:45.114 +08:00] [rocketmq_logger] [---I---] [thread 50329] Sync client settings to ipv4:127.0.0.1:9876 /home/mikehuang/Downloads/rocketmq-clients/cpp/source/client/SessionImpl.cpp:35
[2022/11/14-10:25:45.130 +08:00] [rocketmq_logger] [---W---] [thread 50344] Failed to write telemetry command settings {
  client_type: PUSH_CONSUMER
  access_point {
    scheme: IPv4
    addresses {
      host: "127.0.0.1"
      port: 9876
    }
  }
  request_timeout {
    seconds: 3
  }
  subscription {
    group {
      name: "CID_standard_topic_sample"
    }
    subscriptions {
      topic {
        name: "T_TestTopic"
      }
      expression {
        type: TAG
        expression: "*"
      }
    }
    receive_batch_size: 32
    long_polling_timeout {
      seconds: 30
    }
  }
  user_agent {
    language: CPP
    version: "5.0.0"
    platform: "Linux 64-bit"
    hostname: "localhost"
  }
}
 to ipv4:127.0.0.1:9876 /home/mikehuang/Downloads/rocketmq-clients/cpp/source/client/TelemetryBidiReactor.cpp:77
[2022/11/14-10:25:45.131 +08:00] [rocketmq_logger] [---I---] [thread 50344] ipv4:127.0.0.1:9876#fireClose /home/mikehuang/Downloads/rocketmq-clients/cpp/source/client/TelemetryBidiReactor.cpp:297
[2022/11/14-10:25:45.131 +08:00] [rocketmq_logger] [---W---] [thread 50344] ipv4:127.0.0.1:9876#OnDone, status.error_code=14, status.error_message=failed to connect to all addresses; last error: UNKNOWN: ipv4:127.0.0.1:9876: Cannot check peer: missing selected ALPN property., status.error_details= /home/mikehuang/Downloads/rocketmq-clients/cpp/source/client/TelemetryBidiReactor.cpp:331
[2022/11/14-10:25:46.084 +08:00] [rocketmq_logger] [---I---] [thread 50340] Execute task: heartbeat-task. Use-count: 2 /home/mikehuang/Downloads/rocketmq-clients/cpp/source/scheduler/SchedulerImpl.cpp:165
[2022/11/14-10:25:56.084 +08:00] [rocketmq_logger] [---I---] [thread 50341] Execute task: heartbeat-task. Use-count: 2 /home/mikehuang/Downloads/rocketmq-clients/cpp/source/scheduler/SchedulerImpl.cpp:165

Adjust Java client examples location

client package example in client dir. if package jar for client will be packeged example in. can create new module for examples?

Java: adapt to the latest protocol

Implement native php client

MessageId codec is not using UTC+0

rocketmq-clients/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageIdCodec.java

Line 107 in 62d0deb

final Calendar calendar = Calendar.getInstance();

                    process id(lower 2bytes)
                            ▲
mac address(lower 6bytes)   │   sequence number(big endian)
                   ▲        │          ▲ (4bytes)
                   │        │          │
             ┌─────┴─────┐ ┌┴┐ ┌───┐ ┌─┴─┐
      0x01+  │     6     │ │2│ │ 4 │ │ 4 │
             └───────────┘ └─┘ └─┬─┘ └───┘
                                 │
                                 ▼
          seconds since 2021-01-01 00:00:00(UTC+0)
                        (lower 4bytes)

2021-01-01 00:00:00(UTC+0) timestamp is 1609459200

Golang: Make sure 'syncsettings()' succeeds before the client startup

Make sure 'syncsettings()' succeeds before the client startup

java.lang.NoSuchMethodError: 'java.lang.String javax.annotation.Resource.lookup()'

Caused by: java.lang.NoSuchMethodError: 'java.lang.String javax.annotation.Resource.lookup()'
	at org.springframework.context.annotation.CommonAnnotationBeanPostProcessor$ResourceElement.<init>(CommonAnnotationBeanPostProcessor.java:664)
	at org.springframework.context.annotation.CommonAnnotationBeanPostProcessor.lambda$buildResourceMetadata$0(CommonAnnotationBeanPostProcessor.java:395)
	at org.springframework.util.ReflectionUtils.doWithLocalFields(ReflectionUtils.java:677)
	at org.springframework.context.annotation.CommonAnnotationBeanPostProcessor.buildResourceMetadata(CommonAnnotationBeanPostProcessor.java:377)
	at org.springframework.context.annotation.CommonAnnotationBeanPostProcessor.findResourceMetadata(CommonAnnotationBeanPostProcessor.java:358)
	at org.springframework.context.annotation.CommonAnnotationBeanPostProcessor.postProcessMergedBeanDefinition(CommonAnnotationBeanPostProcessor.java:306)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyMergedBeanDefinitionPostProcessors(AbstractAutowireCapableBeanFactory.java:1116)
	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:594)

RocketMQ(Java) Version: 5.0.0
Spring Version: 5.3.22
Spring Boot Version: 2.7.3

Add issue templates

golang invisibleDuration cause slow consumption for the first time

For the first time, need to wait context_deadline, and timeout = invidibleDuration+timeout.

The roadmap of the golang client

Since the initial implementation of the golang client has been merged(#19), there are still some unfinished tasks to be done.

In the first stage, the basic features will be implemented.

• Adapt to v2 protocol
  The v1 version is archived. It is necessary to adapt to the latest protocol.
• Adapt to the unified API specs ( Producer + Simple Consumer )
  We established a New and Unified API Specs, see more details from java apis. The part of producer and simple consumer takes precedence. The four types of message will be supported in this phase.

In the second stage.

• Clear error handling
  The RPC-status code mapping table indicates the status codes that may occur in various cases across languages, you can get more information from this.
• Improve the observability
  The observability contains logging, tracing and metrics, you can improve observability of client-side through OpenTelemetry or other tools from these aspects.
• Adapt to the unified API specs ( Push Consumer )
  Implement the push consumer.

Documentations and unit tests should cover the corresponding part in each phase.

Doc: List of gRPC header fields

Golang: adapt to the unified API specs

We established a New and Unified API Specs. Maybe it is time for the golang client to adapt to the specs.

Note: The part of producer and simple consumer takes precedence.

Refer to: java apis. Related issue: #29

Golang: Make each 'Client' have an independent 'ClientManager'

Currently a process globally shares a 'ClientManager', so only one 'Conn' is established for each endpoint. When users create a new client, their expectations should be completely independent.

Golang: Somethings todo in the near future

Currently the golang sdk has submitted the first beta version(#160 ), there are still some things todo

#166 GenMessageId
#167 onVerifyMessageCommand
#168 Supports the ability to upload span to the user's server
#169 PushConsumer mode
#170 PrintThreadStackTrace
#171 Grpc connection failed

Golang: Supports the ability to upload span to the user's server

To use opencensue to implement tracing capability, first provide the external interface registerTracingSampler to allow users to pass in their own provider; and report tracing data when a message occurs (refer to the implementation of c++ sdk)

Add copyright xml file for java client

Add copyright xml file for java client

Setup PHP CI pipeline

Golang: PrintThreadStackTrace

When the server sends a request to print the stack, the stack information needs to be reported, and the part of obtaining the stack information needs to be implemented in the onPrintThreadStackTraceCommand function.

Question ：Failed ALPN negotiation: Unable to find compatible protocol

I need some help，
I tried to connect to my local RocketMQ service, but this exception left me confused

Here's my code ，thanks

    public static void main(String[] args){
        ClientServiceProvider provider = ClientServiceProvider.loadService();

        ClientConfiguration configuration = ClientConfiguration.newBuilder()
                .setRequestTimeout(Duration.ofSeconds(10))
                .setEndpoints("172.18.98.74:9876")
                .build();
        try {
            Producer producer = provider.newProducerBuilder()
                    .setClientConfiguration(configuration)
                    .build();

            Message message = provider.newMessageBuilder()
                    .setTopic("TOPIC_TEST")
                    .setBody("HelloWorld".getBytes())
                    .build();

            producer.send(message);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }

Console message

Exception in thread "main" io.grpc.StatusRuntimeException: UNAVAILABLE: Failed ALPN negotiation: Unable to find compatible protocol
Channel Pipeline: [SslHandler#0, ProtocolNegotiators$ClientTlsHandler#0, WriteBufferingAndExceptionHandler#0, DefaultChannelPipeline$TailContext#0]
	at io.grpc.Status.asRuntimeException(Status.java:539)
	at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:544)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:563)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:744)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:723)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)

Announcement for the release of rocketmq-client-java v5.0.1

Golang: PushConsumer mode

Also need to support pushconsumer mode in golang.

The specific design and process are as follows.
https://github.com/apache/rocketmq-clients/blob/master/docs/workflow.md

Doc: add retry policy when flow control occurs

The blocking call in RpcClientImpl#receiveMessage may exhaust all async worker threads.

The default num of shared async worker thread in RpcClientImpl equals to the CPU number. The subsequent task could not be executed in time if the quantity of long-polling request exceeds the CPU num.

Are there plans to support multiple runtimes for rust client?

Add code example for java client

To help newcomers get started easily, we may need more code example.

Golang: How to add Properties

Remove MessageBuilder#setTraceContext and MessageView#getTraceContext

Since propagate trace context by thread local is more general, we should remove MessageBuilder#setTraceContext and MessageView#getTraceContext because propagate context by the serialization is tricky.

Java: refactor MessageMeter

Golang: Print requestId to log

When the status in the response obtained by grpc.invoke is not 200, get the requestId and print it in the log (requestId is obtained from the header)

Golang: Grpc connection failed

A probabilistic error occurs when using the official Grpc-go library to connect to the rocketmq server. The reason is that http2 connection establishment failed. Retry is currently used to reduce this exception, and better solutions can be further explored.

Add a golang client

Add Apache License checker workflow

endpoints doesn't support http?

does endpoints support http url.

Docs: metrics of client

Shutdown ClientMeterManager during the closing of producer/consumer

Connection leak will occur if we don't shutdown the ClientMeterManager.

[GSoC] Integrate with Spring

Similar to the implementation of remoting-based clients in rocketmq-spring, we also need to integrate the rocketmq 5.0 client with Spring.

Golang: Try to recover the When the observer is release

When the observer is shut down, try to recover (after the next routing information update, check if the endpoints still exist, and then call the telemeter function)

Golang: adapt to the v2 protocol

The v1 version is archived. It is necessary to adapt to the latest protocol for the golang client.

related issue: #29