iComposer & AI

iComposer

iComposer Message Management

Overview

iComposer Message Management is a distributed transactional messaging framework that enables asynchronous, reliable message delivery and execution within iComposer orchestration flows. Key capabilities include:

Asynchronous message scheduling and execution
Automatic retry and failure recovery
Distributed locking to ensure single-node processing in clustered environments
TCC transaction support (Seata integration)
Automatic archiving of completed messages to a history table
REST API for message querying, manual retry, and status reset

Architecture Overview

┌──────────────┐     TCC prepare      ┌──────────────────────┐
│  Business    │ ─────────────────►   │  T_ICMP_TX_MESSAGE   │  (current table)
│  Code        │                      │  status=NON_COMMIT(-1)│
│ (Groovy API) │                      └──────────┬───────────┘
└──────────────┘                                 │ TCC commit → status=INITIAL(0)
                                                 ▼
                                      ┌──────────────────────┐
                                      │ IComposerMessage      │
                                      │ Scheduler (polling)   │
                                      └──────────┬───────────┘
                                                 │ mark DOING(1)
                                                 ▼
                                      ┌──────────────────────┐
                                      │ Thread Pool           │
                                      │ GroovyRestfulExecutor │
                                      └──────────┬───────────┘
                                           ┌─────┴──────┐
                                           ▼            ▼
                                      SUCCESS(2)    ERROR(3)
                                           │            │
                                           │       auto-retry (≤maxRetryTimes)
                                           ▼            ▼
                                      ┌──────────────────────────┐
                                      │ T_ICMP_TX_MESSAGE_HIS    │  (history table)
                                      └──────────────────────────┘

Message Status Lifecycle

Status Code	Name	Description
`-1`	NON_COMMIT	TCC prepare phase; message reserved but not yet committed
`0`	INITIAL	Committed; waiting for scheduler to pick up and execute
`1`	DOING	Currently being processed
`2`	SUCCESS	Executed successfully
`3`	ERROR	Execution failed; awaiting retry
`4`	BLOCK	Blocked; requires manual intervention

Status transition paths:

NON_COMMIT(-1) ──TCC commit──► INITIAL(0) ──scheduler──► DOING(1) ──success──► SUCCESS(2)
                                                      ──fail───► ERROR(3) ──retry──► DOING(1)
                                                                 ──maxRetry──► BLOCK(4)
NON_COMMIT(-1) ──TCC rollback──► deleted
DOING(1) ──manual reset──► INITIAL(0)

Configuration

Configure the following parameters in application.properties or via environment variables:

Parameter	Default	Description
`icomposer.message.enabled`	None (must be explicitly set to `true`)	Enable the message scheduler
`icomposer.message.cron`	`0/5 * * * * ?`	Cron expression for message processing (default: every 5 seconds)
`icomposer.message.backupCron`	`0/10 * * * * ?`	Cron expression for message archiving (default: every 10 seconds)
`icomposer.message.batchSize`	`10`	Number of messages fetched per scheduling cycle
`icomposer.message.schedulerLockTimeoutSeconds`	`30`	Distributed lock timeout in seconds
`icomposer.message.retryIntervalMinutes`	`2`	Retry interval for failed messages in minutes
`icomposer.message.maxRetryTimes`	`5`	Maximum number of retry attempts
`icomposer.message.backupBatchSize`	`500`	Batch size for archiving messages
`icomposer.message.useSystemUser`	`false`	Use system user for execution (`false` = use message creator)

Minimal Configuration

# Enable message scheduler
icomposer.message.enabled=true

Production Tuning Example

icomposer.message.enabled=true
icomposer.message.cron=0/3 * * * * ?
icomposer.message.batchSize=20
icomposer.message.retryIntervalMinutes=5
icomposer.message.maxRetryTimes=10
icomposer.message.backupBatchSize=1000
icomposer.message.useSystemUser=false

REST API

All API paths are prefixed with /v1/flow/message.

1. Query Current Messages (Paginated)

POST /v1/flow/message/pagedQueryCurrent

Query messages from the current message table (T_ICMP_TX_MESSAGE).

Request Body:

{
  "pageIndex": 1,
  "pageSize": 20,
  "businessNo": "POL-2025-001",
  "functionName": "policyService",
  "methodName": "issuePolicy",
  "globalTransactionId": "tx-uuid-xxx",
  "status": "3",
  "traceId": "trace-xxx",
  "submitTimeFrom": "2025-05-01T00:00:00",
  "submitTimeTo": "2025-05-31T23:59:59",
  "processTimeFrom": null,
  "processTimeTo": null
}

Query Parameters:

Field	Type	Required	Description
`pageIndex`	Integer	Yes	Page number (1-based)
`pageSize`	Integer	Yes	Page size
`businessNo`	String	No	Business number (fuzzy match)
`functionName`	String	No	Groovy function name
`methodName`	String	No	Groovy method name
`globalTransactionId`	String	No	Global transaction ID
`status`	String	No	Message status code
`traceId`	String	No	Trace ID
`submitTimeFrom/To`	LocalDateTime	No	Submit time range
`processTimeFrom/To`	LocalDateTime	No	Process time range
`endTimeFrom/To`	LocalDateTime	No	End time range

Response:

{
  "data": [
    {
      "messageId": 1001,
      "globalTransactionId": "tx-uuid-xxx",
      "traceId": "trace-xxx",
      "businessNo": "POL-2025-001",
      "submitTime": "2025-05-20T10:30:00",
      "processTime": "2025-05-20T10:30:05",
      "endTime": null,
      "transactionType": 1,
      "isSuccessful": "N",
      "errorMessage": "Connection timeout",
      "functionName": "policyService",
      "methodName": "issuePolicy",
      "status": "3",
      "redoTimes": 2,
      "lastRedoTime": "2025-05-20T10:35:05"
    }
  ],
  "totalCount": 150,
  "pageIndex": 1,
  "pageSize": 20
}

2. Query Historical Messages (Paginated)

POST /v1/flow/message/pagedQuery

Query messages from the history archive table (T_ICMP_TX_MESSAGE_HIS). The request body and response format are identical to pagedQueryCurrent.

3. Retry Failed Messages

POST /v1/flow/message/redo

Re-queue failed messages from the history table back into the current message table for re-execution.

Request Body:

[1001, 1002, 1003]

A list of hisId values from the history table.

Response:

{
  "code": "0",
  "message": "success"
}

Note: The retry operation extracts message details from the history table, re-inserts them into the current message table with INITIAL(0) status, and the scheduler picks them up automatically.

4. Reset Message Status

POST /v1/flow/message/reset

Reset DOING(1) messages back to INITIAL(0). Used to recover messages stuck in DOING state due to node crashes or other failures.

Request Body:

[2001, 2002]

A list of messageId values from the current message table.

Response:

{
  "code": "0",
  "message": "success"
}

Use case: When messages remain in DOING status for an extended period (e.g., the executing node crashed), use this API to reset them to INITIAL so the scheduler can pick them up again.

Usage in Groovy Code

Use IComposerExecutorUtils.asyncExecuteWithMessage to submit asynchronous messages. This method internally uses the TCC pattern to ensure consistency between the message and the business transaction: the message takes effect when the outer transaction commits, and is automatically deleted on rollback.

Basic Usage

import com.insuremo.icomposer.utils.IComposerExecutorUtils

// Minimal form: businessNo + function + method + params
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo - business number (for querying and tracking)
    "policyService",      // serviceFunction - Groovy function name to invoke
    "issuePolicy",        // methodName - method name to execute
    policyData            // params - method parameters (varargs)
)

Specifying Transaction Type

// With custom txType (transaction type identifier, defaults to 100)
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo
    200L,                 // txType - custom transaction type
    "policyService",      // serviceFunction
    "issuePolicy",        // methodName
    policyData            // params
)

Using Map Parameters

When there are many parameters, you can pass them as a Map (use LinkedHashMap to preserve parameter order):

// Map parameter form
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo
    "policyService",      // serviceFunction
    "issuePolicy",        // methodName
    [
        policyNo: "POL-2025-001",
        productCode: "TRAVEL-PA",
        insuredAmount: 100000
    ]                     // methodParamsMap - LinkedHashMap preserves order
)

// Map parameters + custom txType
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo
    200L,                 // txType
    "policyService",      // serviceFunction
    "issuePolicy",        // methodName
    [policyNo: "POL-2025-001", productCode: "TRAVEL-PA"]
)

Method Signatures

Signature	Description
`asyncExecuteWithMessage(businessNo, serviceFunction, methodName, Object... params)`	Basic form; txType defaults to 100
`asyncExecuteWithMessage(businessNo, txType, serviceFunction, methodName, Object... params)`	Specify transaction type
`asyncExecuteWithMessage(businessNo, serviceFunction, methodName, Map<String, Object> paramsMap)`	Map parameters; txType defaults to 100
`asyncExecuteWithMessage(businessNo, txType, serviceFunction, methodName, Map<String, Object> paramsMap)`	Map parameters + custom transaction type

Tracing Message Execution

A traceId is automatically assigned during message execution. Use the traceId in your logging system to trace the full execution chain.

How the Scheduler Works

Message Processing Flow

Scheduled polling: The scheduler triggers based on the cron configuration (default: every 5 seconds)
Distributed lock: Acquires a non-reentrant lock to ensure only one node processes messages at a time
Fetch pending messages:
- Prioritizes INITIAL(0) status messages
- If fewer than batchSize, supplements with ERROR(3) messages eligible for retry
Mark as executing: Uses optimistic locking to change status from INITIAL/ERROR to DOING
Thread pool execution: Submits messages to the thread pool for asynchronous Groovy function execution
Record result: Updates status to SUCCESS or ERROR after execution, recording traceId and error details

Message Archiving Flow

The scheduler triggers based on backupCron (default: every 10 seconds)
Queries completed messages (SUCCESS and ERROR/BLOCK that have reached the retry limit)
Batch-migrates them to the history table

Operations Guide

Common Scenarios

View Failed Messages

POST /v1/flow/message/pagedQueryCurrent
{
  "pageIndex": 1,
  "pageSize": 50,
  "status": "3"
}

Manually Retry Failed Messages

Find the hisId of messages to retry from the history table
Call the redo API:

POST /v1/flow/message/redo
[1001, 1002]

Reset Stuck DOING Messages

When messages remain in DOING status due to a node crash:

POST /v1/flow/message/reset
[2001, 2002]

Track Messages by Business Number

POST /v1/flow/message/pagedQueryCurrent
{
  "pageIndex": 1,
  "pageSize": 20,
  "businessNo": "POL-2025-001"
}

Monitoring Checklist

Watch for accumulation of ERROR and BLOCK status messages
Monitor thread pool queue size and active thread count
Watch for taskExecutor rejected warnings in scheduler logs
Check if distributed locks are timing out frequently

Important Notes

The scheduler is only activated when icomposer.message.enabled=true
Distributed locking is database-based, relying on the LockUtils utility
Message execution uses the message creator’s user context; falls back to ADMIN user if the user no longer exists
When icomposer.message.useSystemUser=true, all messages execute under the system user identity
Messages that reach maxRetryTimes are not automatically archived and require manual handling via the redo/reset APIs