: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), China (Beijing), China (Zhangjiakou), China (Hohhot), China (Shenzhen), China (Chengdu), China (Hong Kong), 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 ApsaraMQ for RocketMQ.

• You can use the TCP client SDK for Java 2.x. x.Final to access ApsaraMQ for RocketMQ only in virtual private clouds (VPCs).

  If you use an existing ApsaraMQ for RocketMQ instance and access the instance in the classic network, do not upgrade the TCP client SDK for Java to 2.x.x.Final. Otherwise, you cannot access the ApsaraMQ for 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 ApsaraMQ for RocketMQ 5.x instances contain namespaces. If you use an ApsaraMQ for 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 ApsaraMQ for 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 feature

Asynchronous logs are supported.

Fixed issues

• The waiting time for batch consumption cannot be specified.

• Specific 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

More data is returned in trace query results.

Feature changes in V2.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 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 the dead-letter queue. You can change the value of the MaxReconsumeTimes parameter to change the maximum number of retries for ordered messages.

Transactional message

When a producer sends messages, an exception is thrown if the LocalTransactionExecutor 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");
        // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for 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);
        // The frequency 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();
    }
}

Consumption in Push mode

• Batch consumption of normal messages 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 a consumer.

• The consumption throttling feature is added. You can configure the consumption throttling feature to limit message consumption rates. 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 specify a custom 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");
          // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console. 
          properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXX");
  
          Consumer consumer = ONSFactory.createConsumer(properties);
  
          // Specify the message consumption rate in testTopicA. In this example, 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();
      }
  }

Consumption in Pull mode

Consumption in Pull mode is not supported.

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 are supported to add log-related parameters. Earlier versions support only -D.

Client creation

If an invalid endpoint is specified, an exception is thrown when the system attempts to create 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 information about the sample SDK code for STS authorization, see Step 2: Access resources across Alibaba Cloud accounts.