:Release notes for the SDK for Java

Usage notes

• You can upgrade your TCP client SDK for Java to V2.x.x.Final only in the following regions: China (Hangzhou), China (Qingdao), and China (Beijing), China (Zhangjiakou), China (Hohhot), China (Shenzhen), China (Chengdu), Germany (Frankfurt), and Indonesia (Jakarta). Do not upgrade your TCP client SDK for Java to V2.x.x.Final in other regions. Otherwise, you cannot use the SDK to connect to Message Queue for Apache RocketMQ.
• You can use the TCP client SDK for Java 2.x. x.Final to access Message Queue for Apache RocketMQ only over a virtual private cloud (VPC).

  If you use an existing Message Queue for Apache RocketMQ instance and access the instance by using the classic network, do not upgrade the TCP client SDK for Java to 2.x.x.Final. Otherwise, you cannot access the Message Queue for Apache RocketMQ instance.

• You can use the TCP client SDK for Java 2.x.x.Final to access only instances that contain namespaces. If the instance that you use does not contain a namespace, do not upgrade the TCP client SDK for Java to 2.x.x.Final.

  By default, all Message Queue for Apache RocketMQ 5.x instances contain namespaces. If you use a Message Queue for Apache RocketMQ 4.x instance, you can check whether the instance contains a namespace in the Basic Information section of the Instance Details page in the Message Queue for Apache RocketMQ console.

Version information

Environment requirements

If you want to use the SDK for Java V2.x.x, make sure that you use a Java Development Kit (JDK) that supports Java 8 or later.

Feature changes in V2.0.5

Optimized features

Asynchronous logs are supported.

Fixed issues

• The waiting time for batch consumption cannot be specified.
• Some security vulnerabilities are fixed.

Feature changes in V2.0.3

Fixed issue

The number of threads in a thread pool cannot be increased to a value larger than 32 in later JDK versions.

Feature changes in V2.0.2

Fixed issue

A deadlock might be triggered during message sending.

Feature changes in V2.0.1

Message trace

Data about message traces is added.

Feature changes in 2.0.0

Ordered message

The default value of the MaxReconsumeTimes parameter is changed from Integer.MAX to 16. This parameter specifies the maximum number of retries allowed for ordered messages. If a consumer still fails to consume a message after the maximum number of retries is reached, the message is delivered to a dead-letter queue. You can change the value of the MaxReconsumeTimes parameter to change the maximum number of retries for ordered messages.

Transactional message

In V2.0.0, when a producer sends messages, an exception is thrown if the LocalTransactionExecuter class is null. No exception is thrown for this error in earlier versions.

Broadcasting consumption

In broadcasting consumption mode, you can call the offsetStore operation to specify the consumer offset from which consumption starts. If you do not specify the consumer offset, the consumption starts from the latest consumer offset. This is consistent with earlier versions.

public class BroadcastingConsumerExample {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();

        properties.put(PropertyKeyConst.GROUP_ID, "MyGroupId");
        properties.put(PropertyKeyConst.AccessKey, "MyAccessKey");
        properties.put(PropertyKeyConst.SecretKey, "MySecretKey");
        // Specify the endpoint that is used to access the Message Queue for Apache RocketMQ instance over TCP. You can view the endpoint on the Endpoints tab of the Instance Details page in the Message Queue for Apache RocketMQ console. 
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        // You can obtain the consumer offset by calling the offsetStore operations only in broadcasting consumption mode. 
        properties.put(PropertyKeyConst.MessageModel, MessageModel.BROADCASTING);

        Consumer consumer = ONSFactory.createConsumer(properties);
        // Set the number of calls for the AbstractOffsetStore method. The value 1 specifies that the system calls the AbstractOffsetStore method every second to set a persistent offset. 
        OffsetStore offsetStore = new AbstractOffsetStore(1) {
            @Override
            public Map<TopicPartition, Long> loadOffset() {
            // The logic that is used to obtain the offset from an external storage system. 
            }

            @Override
            public void persistOffset(Map<TopicPartition, Long> offsetTable) {
            // The logic that is used to persist the offset to an external storage system. 
            }
        };
        offsetStore.start();
        consumer.setOffsetStore(offsetStore);

        consumer.subscribe("testBroadcastingTopic", "TagA", new MessageListener() {
            @Override
            public Action consume(Message message, ConsumeContext context) {
                // The logic that is used to consume messages. 
                return Action.CommitMessage;
            }
        });

        consumer.start();
        Thread.sleep(100000);
        consumer.shutdown();
        offsetStore.shutdown();
    }
}

Push mode

• Consumption of multiple messages at a time is not supported.
• If the specified number of consumption threads is not within the valid range of 1 to 1,000, the system throws an exception when you create but not start a consumer.
• The consumption throttling feature is added. You can configure the consumption throttling feature to limit the consumption rate of messages. This helps prevent application exceptions that are caused by sudden surges of messages on consumer clients. The following code provides an example on how to configure message consumption rate:
  Note The consumption throttling feature does not apply to the retries of ordered messages.

  public class RateLimitConsumerExample {
      public static void main(String[] args) throws InterruptedException {
          Properties properties = new Properties();
  
          properties.put(PropertyKeyConst.GROUP_ID, "MyGroupId");
          properties.put(PropertyKeyConst.AccessKey, "MyAccessKey");
          properties.put(PropertyKeyConst.SecretKey, "MySecretKey");
          // Specify the endpoint that is used to access the Message Queue for Apache RocketMQ instance over TCP. You can view the endpoint on the Endpoints tab of the Instance Details page in the Message Queue for Apache RocketMQ console. 
          properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXX");
  
          Consumer consumer = ONSFactory.createConsumer(properties);
  
          // Set the consumption rate of messages in testTopicA. The consumer client can consume only 10 messages per second. 
          consumer.rateLimit("testTopicA", 10);
          consumer.subscribe("testTopic", "TagA", new MessageListener() {
              @Override
              public Action consume(Message message, ConsumeContext context) {
                  // The logic that is used to consume messages. 
                  return Action.CommitMessage;
              }
          });
  
          consumer.start();
          Thread.sleep(100000);
          consumer.shutdown();
      }
  }

Pull mode

The pull mode for message consumption is not supported in V2.0.0.

Log configuration

• The default log path is changed from ~/logs/ons.log to ~/logs/ons/ons-client.log.
• Log levels are consistent with the log levels that are supported in the logback framework. The OFF, TRACE, and ALL log levels are added. Earlier versions support only the ERROR, WARN, INFO, and DEBUG log levels.
• Environment variables can be used to add log-related parameters. In earlier versions, log-related parameters can be added only by using the -D option.

Client creation

If an invalid endpoint is specified, an exception is thrown when the system attempts to create but not start a producer or a consumer.

Message trace

Fixed issues in V2.0.0

The following issue is fixed: When the updateCredential operation is called multiple times in a short period of time, cross-account authorization for RAM roles by using Security Token Service (STS) fails because atomicity is not ensured when the AccessKey ID, AccessKey secret, and STS token are updated.

For the sample SDK code for STS authorization, see the configuration example of an STS token in Grant permissions across Alibaba Cloud accounts by using a RAM role.