Message-Related

Last updated: 2025-09-28 16:20:09

Message-Related

Last updated: 2025-09-28 16:20:09

How do I calculate the number of unconsumed messages?
Number of unconsumed messages = maximum offset - submitted offset, as shown below:
img
﻿
Can the message retention period be automatically adjusted?
CKafka allows you to add dynamic message retention policies. After such a policy is set, if the disk utilization reaches the specified percentage, a certain proportion of existing data will automatically expire in case of message surges, helping avoid situations where normal production and consumption stop after the disk space is used up. For detailed directions, please see Adding Dynamic Message Retention Policy.
Why Is Consumption Progress Not Synchronized to the Server in Time
Symptom
CKafka consumption progress is not updated on the server in time, and the consumption speed cannot be seen via console.
Pre-Reserve
The CKafka cluster shows consumption progress based on the commit frequency of the client consumption offset to estimate consumption speed and progress. Therefore, prioritize troubleshooting client consumption and offset submission in such problems.
﻿
Troubleshooting Guide
Step 1: Check Whether There Is Production Data Written
If no accumulation or consumption speed is detected, check the Topic and production monitoring metrics to see if data is being written.
1. Enter the console, click instance list > ID/Name > Topic list, and check whether the end offset of related partitions updates.
﻿
2. Enter the console, click instance list > ID/Name > monitor > Topic, and check whether the production flow and total amount of messages occupying disk exceed 0.
﻿
If no real-time data is written to the Topic, there are two cases:
Create a consumer group, set the auto.offset.reset=latest behavior to start consuming from the latest. Since there is no real-time data write, it will not generate commit offset action;
Topic data exceeds the retention time. If you set the auto.offset.reset=earliest behavior, it will start consuming from the beginning. Since the data has been retention, consumption failure occurs, and no commit offset action will be generated.
The above scenario does not generate commit offset action, so the server will not update the consumption progress.
Step 2: Check Client Configuration
For Kafka native Client, need to check Consumer configuration, whether auto commit offset: enable.auto.commit is enabled.
Native Kafka Java Client Example
A direct way is to check the client configuration, or you can search for ConsumerConfig in the program logs, as shown below. This log is only printed during KafkaConsumer initialization and may be unable to find if the log retention time is short.
﻿
Based on client consumption features, if auto commit offset can be enabled, you can turn on this parameter, and client consumption progress will be updated to the server;
If manual offset submission management is required, such as stricter requirements for duplicate consumption or data loss, when enable.auto.commit is off, check whether the client program synchronously or asynchronously performs offset submission actions during consumption, for example consumer.commitSync()
while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(500);
            for (ConsumerRecord record : consumerRecords) {
                System.out.printf("receive message: partition:%d, offset:%d, time:%d, key:%s, value:%s\\n", record.partition(), record.offset(), record.timestamp(), record.key(), record.value());
            }
            consumer.commitSync();
        }
After troubleshooting and adjustment with the Native Kafka Client, observe again whether Client consumption progress is updated to the server.
Targeting Kafka Streaming Scenarios with Relatively High Flink Occurrence Frequency
For the Flink consumer client, its underlying layer integrates with Kafka through the Flink Kafka Source API, combining with the Flink framework's fault tolerance mechanism, such as checkpoint and savepoint mechanisms. Here, two cases arise: Kafka Client's underlying enable.auto.commit handling and Flink checkpoint's offset processing.
Kafka Source submits the current consumer offset upon checkpoint completion to ensure Flink checkpoint status matches the commit offset on the Kafka broker.
 If checkpoint is not enabled, Kafka Source depends on the Kafka consumer's internal scheduled auto-commit logic for offsets. The autocommit feature is configured by the two Kafka consumer configuration items: enable.auto.commit and auto.commit.interval.ms. Notably, If the user does not manually specify enable.auto.commit in the configuration when using Flink Kafka Source, the Flink framework will override enable.auto.commit=false, disabling auto-commit of offsets. If Flink checkpoint is not enabled, the user is required to manually enable this parameter.
Expired Messages Not Promptly Deleted What Is the Cause
Cause Analysis
The message deletion mechanism in Kafka can cause expired messages to not be promptly deleted in specific business scenarios. Lack of understanding of the mechanism may lead to doubt. For example, a significant gap exists between message timestamps in partition 0 and partition 7, with expired messages in partition 0 not being deleted in time.
Kafka Message Deletion Mechanism
Kafka data storage is physically stored on disk in three dimensions: Topic, partition, and data segment. Message data deletion conditions are as follows:
Message data is deleted based on retention time, with deletion performed at the data segment level.
Each data segment is set to 1GB in size. When it reaches 1GB, a new segment is generated by rolling.
All messages in a data segment must expire before the segment is deleted.
If a message in the data segment is within the retention time, such as the last line of the segment file, this segment file will not be deleted.
Due to some reasons, message write is skewed, and data ingestion is concentrated in specific partitions, such as partition 7, while data for some partitions is seldom, such as partition 0. At this point, the size of the data segment in partition 0 does not reach 1GB, so no scrolling occurs. However, since there is data within the retention time in the entire segment, messages in partition 0 will not be deleted.
﻿