Lite topic model

Lite topic name

• Definition: The name that identifies a lite topic. Lite topic names are globally unique within their parent topic.

• Value: When the topic type is Lite and you call setLiteTopic on a message, the system automatically creates the lite topic if it does not already exist.

• Constraint: See Parameter limits.

Time-to-live (TTL)

• Definition: The expiration time for a lite topic. If no new messages are written to the lite topic for longer than its TTL, the system automatically deletes it. Deletion means releasing the count allocated to the lite topic (total count minus one).

• Value: When creating a Lite-type topic, you can set the expiration parameter.

• Constraint: See Parameter limits.

Use case 1: Asynchronous communication for Multi-Agent systems to resolve long-running call blocking

As AI scenarios grow more complex, single-agent systems face limitations: lack of specialization, difficulty integrating multiple domains, and inability to enable dynamic collaborative decision-making. Single-agent applications and workflows are shifting toward Multi-Agent architectures. However, because AI tasks often take a long time, synchronous calls block the caller’s thread, limiting scalability for large-scale collaboration.

As shown above, the Multi-Agent workflow works like this: The Supervisor Agent splits a request into two subtasks for two child agents. Each child agent solves its part and returns results to the Supervisor Agent, which aggregates them and sends the final response to the web client. Using RocketMQ for asynchronous communication:

1. Request handling flow:

   1. Create a topic (Request) for each child agent as a task buffer queue. Use a priority topic to process high-priority tasks first.

   2. The Supervisor Agent sends split task details to the corresponding request topic.

2. Response handling flow:

   1. The Supervisor Agent creates a Lite-type topic (Response) and subscribes to it.

   2. Each child agent sends its task result to a LiteTopic under the Response topic. Name each LiteTopic using the task ID to give every task its own dedicated LiteTopic.

   3. The Supervisor Agent receives results in real time through subscription and pushes them to the web client using HTTP SSE.

Use case 2: Distributed session state management to solve session continuity issues in AI applications

AI application interactions are unique: long-running, multi-turn, and heavily dependent on expensive compute resources per session. When applications rely on persistent connections such as SSE, any disconnection (due to gateway restarts, timeouts, or network instability) causes loss of current session context and wastes already invested AI compute resources.

As in use case 1’s response flow, use a Lite-type topic for real-time result notifications. Name each LiteTopic using the SessionID (for example, chatbot/{sessionID}). All session results are delivered as ordered messages in this topic. To maintain session continuity after reconnection:

1. The web client establishes a persistent connection with application server node 1 and starts session Session2.

2. Application server node 1 subscribes to LiteTopic [chat/SessionID2].

3. The Large Language Model (LLM) task scheduler sends results to LiteTopic [chat/SessionID2] based on the SessionID in the request.

4. Due to a network issue, the WebSocket reconnects to application server node 2.

5. Application server node 1 unsubscribes from LiteTopic [chat/SessionID2]. Application server node 2 subscribes to it.

6. LiteTopic [chat/SessionID2] resumes delivery from the last consumed offset, ensuring continuous session state and data.

Send messages

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();
final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // Set a message key for precise lookup by keyword.
    .setKeys("messageKey")
    // Set LiteTopic
    .setLiteTopic("lite-topic-1")
    // Message body
    .setBody("messageBody".getBytes())
    .build();
try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // LiteTopic quota exceeded. Evaluate and increase quota.
    log.error("Lite topic quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to send message", t);
}

Consume messages

Use the LitePushConsumer class:

// Initialize LitePushConsumer with consumer group, target topic, and communication parameters.
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // Topic bound to the ConsumerGroup when created in the console
    .bindTopic(topicName)
    // Set consumer group
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // Process message and return consumption result.
        LOGGER.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();
try {
    // Subscribe to desired LiteTopics
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // LiteTopic subscription quota exceeded. Evaluate and increase quota.
    log.error("Lite subscription quota exceeded", e);
} catch (Throwable t) {
    log.error("Failed to subscribe lite topic", t);
}
// After business processing, unsubscribe from unused LiteTopics promptly
litePushConsumer.unsubscribeLite("lite-topic-3");
// Get current set of subscribed LiteTopics
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

Dynamically update subscriptions

/**
 * Dynamically add a subscription.
 * The subscribeLite() method makes a network call and validates quotas,
 * so it might fail.
 * Always check the result to confirm successful subscription.
 * Possible failure scenarios:
 * 1. Network error – retry the call.
 * 2. Quota validation fails – throws LiteSubscriptionQuotaExceededException.
 * Evaluate whether your quota meets requirements, and promptly call
 * unsubscribeLite() to release resources for unused topics.
 */
litePushConsumer.subscribeLite("lite-topic-1");
// Dynamically remove a subscription
litePushConsumer.unsubscribeLite("lite-topic-1");

Serverless instances

Non-Serverless instances (subscription and pay-as-you-go)