[ISSUE #1107] [Java] LiteTopic support for "RIP‐83 Lite Topic" (#1108)

* producer lite topic

* new MessageImpl with MessageBuilderImpl

* lite push consumer

* createFifoConsumeService

* default fifo true for LitePushConsumerSettings

* lite push consumer sub bindTopic *

* 1. fix heartbeat;
2. add version for interest;

* upgrade to 5.0.9-lite-topic-SNAPSHOT and fix logback

* add some test

* subscribeLite List<String>

* Collection instead of List

* check point

* syncLiteSubscription

* code refactor

* fix check style

* onNotifyUnsubscribeLiteCommand

* LiteSubscriptionQuotaExceededException and LiteTopicQuotaExceededException

* subscribeLite atomic

* client side quota check

* client-side lite subscription quota limit

* checkLiteSubscriptionQuota 3000

* validateLiteTopic for lite push consumer

* ack with lite topic if present.

* forward syncSettings exception to upper layer

* set liteSubscriptionQuota 1200

* lite push conusmer setInvisibleDuration

* LitePushConsumerSettings#toString more fields

* revert InvisibleDuration

* log when subscribeLite raise ClientException

* LitePushConsumerImplTest

* more clearer LitePushConsumer Interface

* remove subscribeLite(Collection<String>)

* code refactor and more unit test

* remove setEnableFifoConsumeAccelerator

* code recover

* fix

* fix

* fix code style

* add liteTopic for ForwardMessageToDeadLetterQueueRequest

* fix normal group consume lite topic

* log client type

* notify unsubscribe lite with liteTopic only

* fix test

* complete lite example

* LitePushConsumer#getLiteTopicSet

* fix comments

* remove fifo = true

* log warn for onNotifyUnsubscribeLiteCommand default implementation

* use new proto and update to 5.1.0-lite-topic-SNAPSHOT

Change-Id: I01f0c9c64be2614d0215560240548bba2078e505

* log when subscribeLite and unsubscribeLite

Change-Id: I01cb9cd4d9a474218e674662407708cf069bf426

* revert pom config

Change-Id: I7cc3f5624bf73b8cfeb902f4e0fc3b1fc7d4062d

* rocketmq-proto.version to 2.1.0-SNAPSHOT

Change-Id: I686e9ab7eaed1af1d1b01a57a88438af4bfa1878

* liteSubscriptionQuota default to 0

Change-Id: I3bcfaaa323b1eea3bd0665ab1d8722a7bc47a4fe

* upgrade to 5.1.0-SNAPSHOT

Change-Id: I360dbd38de8fa2cc21ab4ebb827fcbdce0258fa6

* add comments

Change-Id: I062b4a92032788b6c361304d45b6a40e9a28923a

* add comments

Change-Id: I53a21ec7faa0ba85d46c020a1709f4e011c799ec

---------

Co-authored-by: Quan <[email protected]>
Co-authored-by: moling <[email protected]>

Author: 3 people

java/client-apis/pom.xml

@@ -18,7 +18,7 @@
     <parent>
         <artifactId>rocketmq-client-java-parent</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.0.9-SNAPSHOT</version>
+        <version>5.1.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientServiceProvider.java

@@ -19,6 +19,7 @@
 
 import java.util.Iterator;
 import java.util.ServiceLoader;
+import org.apache.rocketmq.client.apis.consumer.LitePushConsumerBuilder;
 import org.apache.rocketmq.client.apis.consumer.PushConsumerBuilder;
 import org.apache.rocketmq.client.apis.consumer.SimpleConsumerBuilder;
 import org.apache.rocketmq.client.apis.message.MessageBuilder;
@@ -59,6 +60,13 @@ static ClientServiceProvider loadService() {
      */
     PushConsumerBuilder newPushConsumerBuilder();
 
+    /**
+     * Get the lite push consumer builder by the current provider.
+     *
+     * @return the lite push consumer builder instance.
+     */
+    LitePushConsumerBuilder newLitePushConsumerBuilder();
+
     /**
      * Get the simple consumer builder by the current provider.
      *

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/LitePushConsumer.java

@@ -0,0 +1,73 @@
+/*
+ * 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 org.apache.rocketmq.client.apis.consumer;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Set;
+import org.apache.rocketmq.client.apis.ClientException;
+
+public interface LitePushConsumer extends Closeable {
+
+    /**
+     * Subscribe to a lite topic.
+     * <p>
+     * The subscribeLite() method initiates network requests and performs quota verification, so it may fail.
+     * It's important to check the result of this call to ensure that the subscription was successfully added.
+     * Possible failure scenarios include:
+     * 1. Network request errors, which can be retried.
+     * 2. Quota verification failures, indicated by LiteSubscriptionQuotaExceededException. In this case,
+     *    evaluate whether the quota is insufficient and promptly unsubscribe from unused subscriptions
+     *    using unsubscribeLite() to free up resources.
+     *
+     * @param liteTopic the name of the lite topic to subscribe to
+     * @throws ClientException if an error occurs during subscription
+     */
+    void subscribeLite(String liteTopic) throws ClientException;
+
+    /**
+     * Unsubscribe from a lite topic.
+     *
+     * @param liteTopic the name of the lite topic to unsubscribe from
+     * @throws ClientException if an error occurs during unsubscription
+     */
+    void unsubscribeLite(String liteTopic) throws ClientException;
+
+    /**
+     * Get the lite topic immutable set.
+     *
+     * @return lite topic immutable set.
+     */
+    Set<String> getLiteTopicSet();
+
+    /**
+     * Get the load balancing group for the consumer.
+     *
+     * @return consumer load balancing group.
+     */
+    String getConsumerGroup();
+
+    /**
+     * Close the consumer and release all related resources.
+     *
+     * <p>Once consumer is closed, <strong>it could not be started once again.</strong> we maintained an FSM
+     * (finite-state machine) to record the different states for each push consumer.
+     */
+    @Override
+    void close() throws IOException;
+}

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/LitePushConsumerBuilder.java

@@ -0,0 +1,93 @@
+/*
+ * 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 org.apache.rocketmq.client.apis.consumer;
+
+import org.apache.rocketmq.client.apis.ClientConfiguration;
+import org.apache.rocketmq.client.apis.ClientException;
+
+/**
+ * Builder to config and start {@link LitePushConsumer}.
+ */
+public interface LitePushConsumerBuilder {
+
+    /**
+     * Set the bind topic for lite push consumer.
+     *
+     * @return the consumer builder instance.
+     */
+    LitePushConsumerBuilder bindTopic(String bindTopic);
+
+    /**
+     * Set the client configuration for the consumer.
+     *
+     * @param clientConfiguration client's configuration.
+     * @return the consumer builder instance.
+     */
+    LitePushConsumerBuilder setClientConfiguration(ClientConfiguration clientConfiguration);
+
+    /**
+     * Set the load balancing group for the consumer.
+     *
+     * @param consumerGroup consumer load balancing group.
+     * @return the consumer builder instance.
+     */
+    LitePushConsumerBuilder setConsumerGroup(String consumerGroup);
+
+    /**
+     * Register message listener, all messages meet the subscription expression would across listener here.
+     *
+     * @param listener message listener.
+     * @return the consumer builder instance.
+     */
+    LitePushConsumerBuilder setMessageListener(MessageListener listener);
+
+    /**
+     * Set the maximum number of messages cached locally.
+     *
+     * @param count message count.
+     * @return the consumer builder instance.
+     */
+    LitePushConsumerBuilder setMaxCacheMessageCount(int count);
+
+    /**
+     * Set the maximum bytes of messages cached locally.
+     *
+     * @param bytes message size.
+     * @return the consumer builder instance.
+     */
+    LitePushConsumerBuilder setMaxCacheMessageSizeInBytes(int bytes);
+
+    /**
+     * Set the consumption thread count in parallel.
+     *
+     * @param count thread count.
+     * @return the consumer builder instance.
+     */
+    LitePushConsumerBuilder setConsumptionThreadCount(int count);
+
+    /**
+     * Finalize the build of {@link LitePushConsumer} and start.
+     *
+     * <p>This method will block until the push consumer starts successfully.
+     *
+     * <p>Especially, if this method is invoked more than once, different push consumers will be created and started.
+     *
+     * @return the lite push consumer instance.
+     */
+    LitePushConsumer build() throws ClientException;
+}

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/Message.java

@@ -72,6 +72,13 @@ public interface Message {
      */
     Optional<String> getMessageGroup();
 
+    /**
+     * Get the lite topic, which is used for lite topic message type.
+     *
+     * @return lite topic, which is optional, {@link Optional#empty()} means lite topic is not specified.
+     */
+    Optional<String> getLiteTopic();
+
     /**
      * Get the expected delivery timestamp, which make sense only when topic type is delay.
      *

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageBuilder.java

@@ -87,6 +87,14 @@ public interface MessageBuilder {
      */
     MessageBuilder setMessageGroup(String messageGroup);
 
+    /**
+     * Set the lite topic for the message, which is optional.
+     *
+     * @param liteTopic lite topic for the message.
+     * @return the message builder instance.
+     */
+    MessageBuilder setLiteTopic(String liteTopic);
+
     /**
      * Set the delivery timestamp for the message, which is optional.
      *

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java

@@ -79,6 +79,13 @@ public interface MessageView {
      */
     Optional<String> getMessageGroup();
 
+    /**
+     * Get the lite topic, which makes sense only when the topic type is LITE.
+     *
+     * @return lite topic, which is optional, {@link Optional#empty()} means lite topic is not specified.
+     */
+    Optional<String> getLiteTopic();
+
     /**
      * Get the expected delivery timestamp, which makes sense only when the topic type is delay.
      *

java/client-shade/pom.xml

@@ -18,7 +18,7 @@
     <parent>
         <artifactId>rocketmq-client-java-parent</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.0.9-SNAPSHOT</version>
+        <version>5.1.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

java/client/pom.xml

@@ -18,7 +18,7 @@
     <parent>
         <artifactId>rocketmq-client-java-parent</artifactId>
         <groupId>org.apache.rocketmq</groupId>
-        <version>5.0.9-SNAPSHOT</version>
+        <version>5.1.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

java/client/src/main/java/org/apache/rocketmq/client/java/example/LiteProducerExample.java

@@ -0,0 +1,66 @@
+/*
+ * 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 org.apache.rocketmq.client.java.example;
+
+import java.nio.charset.StandardCharsets;
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.apis.ClientServiceProvider;
+import org.apache.rocketmq.client.apis.message.Message;
+import org.apache.rocketmq.client.apis.producer.Producer;
+import org.apache.rocketmq.client.apis.producer.SendReceipt;
+import org.apache.rocketmq.client.java.exception.LiteTopicQuotaExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LiteProducerExample {
+    static final Logger log = LoggerFactory.getLogger(LiteProducerExample.class);
+
+    private LiteProducerExample() {
+    }
+
+    public static void main(String[] args) throws ClientException {
+        final ClientServiceProvider provider = ClientServiceProvider.loadService();
+
+        String topic = "yourParentTopic";
+        final Producer producer = ProducerSingleton.getInstance(topic);
+        // Define your message body.
+        byte[] body = "This is a lite message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
+        final Message message = provider.newMessageBuilder()
+            // Set topic for the current message.
+            .setTopic(topic)
+            // Key(s) of the message, another way to mark message besides message id.
+            .setKeys("yourMessageKey-3ee439f945d7")
+            // Set your lite topic
+            .setLiteTopic("lite-topic-1")
+            .setBody(body)
+            .build();
+        try {
+            final SendReceipt sendReceipt = producer.send(message);
+            log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
+        } catch (LiteTopicQuotaExceededException e) {
+            // Lite topic quota exceeded.
+            // Evaluate and increase the lite topic resource limit.
+            log.error("Lite topic quota exceeded", e);
+        } catch (Throwable t) {
+            log.error("Failed to send message", t);
+        }
+        // Close the producer when you don't need it anymore.
+        // You could close it manually or add this into the JVM shutdown hook.
+        // producer.close();
+    }
+}