Message delivery - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

Message delivery streams startup logs, resource usage, running events, and job status changes from Flink jobs in a target namespace to Simple Log Service (SLS) or ApsaraMQ for Kafka in real time. By persisting this data, you can query historical records for troubleshooting, performance optimization, and auditing. This topic describes how to configure message delivery and view the delivered messages.

Background

Message delivery supports delivery to SLS and ApsaraMQ for Kafka. Both channels deliver the same message content. SLS supports cross-region delivery. Kafka delivers messages only to an instance in the same region as your Flink workspace. The following table describes the types of information delivered and their delivery triggers.

Category	Description	Delivery trigger
Job startup logs	Logs generated during the entire job startup process, from the initialization of the Flink environment to the startup of the JobManager and the generation of the Flink execution graph.	Delivered once when the job starts successfully or reaches a final state (such as failed or finished).
Resource information	Changes in resource usage within a namespace, including the consumption and allocation of CPU and memory. Changes in resource usage for a queue, including the consumption and allocation of CPU and memory. Important This information is for resource capacity tracking only and does not provide alerts.	Delivered every 30 seconds while the namespace is operating normally.
Runtime events	The startup status at each stage of the job startup process.	Triggered when a job runtime event occurs.
Job resource consumption	Resource usage of running streaming jobs. This does not include batch jobs or jobs running on a session cluster.	Delivered every 10 minutes while the job is running.
Job state change	Job state change events, including the previous and new states.	Triggered when the job state changes.

Notes

Before you deliver messages to SLS, create an SLS project and a Logstore. For instructions, see Use LoongCollector to collect and analyze text logs on ECS instances.
Before you deliver messages to Kafka, create an ApsaraMQ for Kafka instance and a topic in the ApsaraMQ for Kafka console. The topic is not created automatically. For more information, see Create Kafka resources.
SLS delivery and Kafka delivery are independent. You can enable one or both channels, and configure the delivery range for each tab separately.
Message delivery is free of charge. However, you are charged for using SLS features such as Logstore indexing. For more information, see Billing overview.
Querying and analyzing logs in SLS requires an index. Enabling an index is optional and incurs fees for index traffic and storage. For more information about the fees, see Billing overview.
You can configure server-side encryption for the Logstore. Messages delivered to the Logstore are also encrypted. To learn how, see Data encryption.
Message delivery covers startup logs, resource details, full running events, job resource usage, and job status changes. To output job runtime logs, see Configure job log output.
Changes to the message delivery configuration can take up to 30 seconds to take effect.

Procedure

Step 1: Configure message delivery

Go to the message delivery configuration page
1. Log on to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. In the left-side navigation pane, click O&M > Configurations.
4. Click the Message Delivery tab.

Configure message delivery parameters

Complete the configuration on the tab of the target channel.

SLS Delivery

Turn on the Deliver to SLS switch.

Configure the Log Service parameters.

Parameter	Description
Authorization mode	STS Token: This mode delivers messages only to a Log Service project in the same region as your Flink workspace. Therefore, you only need to configure the Log Service project and Logstore. AccessKey: This mode allows message delivery to a Log Service project in a different region. Therefore, you must configure the region, endpoint, AccessKey ID, and AccessKey Secret.
Log Service project	The name of the Log Service project.
Log Service Logstore	The name of the Log Service Logstore.
Region	The region in which the Log Service project resides. If Authorization mode is set to AccessKey, you must configure this parameter. For the regions and endpoints, see Endpoints.
endpoint	The Log Service endpoint. If Authorization mode is set to STS Token, the system automatically uses the endpoint for the region where your Flink workspace is located. If you set Authorization mode to AccessKey, you must manually configure this parameter.
Delivery range	The message content to deliver. For more information, see Field reference later in this topic.
AccessKey ID	Your Alibaba Cloud account's AccessKey ID and AccessKey Secret. Important To prevent your AccessKey credentials from being exposed, we recommend that you manage your AccessKey ID and AccessKey Secret as variables. You can click the drop-down arrow to select an existing variable, or click the icon next to the parameter to create a new variable. For more information about how to manage variables and view your AccessKey ID and AccessKey Secret, see Variable management and How do I view AccessKey credentials?
AccessKey Secret

Click Save.

Kafka Delivery

Turn on the Deliver to Kafka switch.

Configure the Kafka parameters.

Parameter	Description
Authorization Mode	STS Token: delivers messages only to a Kafka instance in the same region as your Flink workspace. Kafka instance, Endpoint, and Kafka Topic are all selected from drop-down lists. Manual configuration: enter Endpoint and Kafka Topic manually.
Kafka instance	The Kafka instance that receives the messages. Displayed when the authorization mode is STS Token. Click the refresh icon to reload the instance list.
Endpoint	The endpoint of the Kafka instance. Automatically filled based on the selected instance when the authorization mode is STS Token. Manually entered when the authorization mode is Manual configuration.
Kafka Topic	The topic that receives the messages. Create the topic in advance. Click Kafka console to go to the ApsaraMQ for Kafka console. When the authorization mode is Manual configuration, the topic name must be 3 to 64 characters in length and can contain only letters, digits, periods (.), hyphens (-), and underscores (_). The name must contain at least one letter or digit.
Delivery range	The message types to deliver. Select at least one of Startup Logs, Resource Details, Full Running Events, Job Resource Usage, and Job Status Changed. For more information about the message content, see Field descriptions later in this topic.

Click Save.

Step 2: View delivered messages

Procedure

Deliver to SLS

Click Open the SLS console to the right of Log Service project.
View the information in the raw logs.

The raw logs are JSON-format messages that contain resource quota information (RESOURCE_QUOTA). These logs include fields such as namespaceTotalCpu (20.0), namespaceTotalMemory (0.08 TiB), namespaceUsedCpu (8.0), namespaceUsedMemory (23.00 GiB), and resourceQueueName (default-queue). These fields record the CPU and memory quotas and usage for the namespace and resource queue.

Deliver to Kafka

Messages are delivered in JSON format to the Kafka topic that you configured in Step 1.
Log on to the ApsaraMQ for Kafka console and view the messages on the Message query page of the target instance. For more information, see Query messages. You can also use a self-managed Kafka client to consume the topic over the configured Endpoint.

Field descriptions

The message body structure is the same for SLS and Kafka. The following sections describe the fields of each message type.

Startup logs (JOB_START_LOG)

Parameter	Description
messageType	Message type. Fixed to JOB_START_LOG (job startup log).
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
tag	The job tag. This field is empty if no tag is configured.
length	The total length of the log.
offset	For sharded logs, this field specifies the start offset of the log entry.
content	Details of the job startup log.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Resource details (JOB_RESOURCE_QUOTA)

Parameter	Description
messageType	Message type. Fixed to RESOURCE_QUOTA (resource usage).
namespaceTotalCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
namespaceTotalCpu	The total CPU resources for the namespace, measured in CUs.
namespaceTotalMemory	The total memory resources for the namespace.
namespaceUsedCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
namespaceUsedCpu	The CPU resources used by the namespace, measured in CUs.
namespaceUsedMemory	The memory resources used by the namespace.
resourceQueueName	The name of the resource queue.
resourceQueueTotalCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
resourceQueueTotalCpu	The total CPU resources for the resource queue, measured in CUs.
resourceQueueTotalMemory	The total memory resources for the resource queue.
resourceQueueUsedCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
resourceQueueUsedCpu	The CPU resources used by the resource queue, measured in CUs.
resourceQueueUsedMemory	The memory resources used by the resource queue.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Runtime events (JOB_EVENT)

Parameter	Description
messageType	Message type. Fixed to JOB_EVENT (runtime event).
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
tag	The job tag. This field is empty if no tag is configured.
eventId	The ID of the event.
eventName	The name of the event.
content	Details of the event.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Job resource usage (JOB_RESOURCE_USAGE)

Parameter	Description
messageType	Message type. Fixed to JOB_RESOURCE_USAGE (job resource usage).
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
tag	The job tag. This field is empty if no tag is configured.
jobUsedCpu	The number of CUs used by the job.
jobUsedMemory	The amount of memory used by the job.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Job status changes (JOB_STATUS_CHANGED)

Parameter	Description
messageType	Message type. Fixed to JOB_STATUS_CHANGED (job status change).
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated, in milliseconds.
workspace	The ID of the workspace.
namespace	The name of the namespace.
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
previousStatus	The job status before the change.
currentStatus	The job status after the change.
tag	The job tag. This field is empty if no tag is configured.
additionalInfo	Additional information.

References

To configure logging for a single job, see Configure job log output.
To view logs in the Realtime Compute for Apache Flink development console, see View startup and running logs, View running events, View running exception logs, and View historical job instance logs.