Hi all, @Rocky, quick question about Routine Load behavior in StarRocks 4.0.7. I stopped a Routine Load job to do a schema change on the target table (dropped some columns, added some new ones), waited until the schema updates were fully done, and then created a new Routine Load job using the same Kafka consumer group name. I expected it to continue from the previous committed offsets / existing lag, but instead it started from the newest Kafka messages and skipped the lag that was still there. Is this expected behavior when recreating a Routine Load job, or should it have resumed from the old offsets? If this is expected, what’s the right way to preserve offsets across this kind of schema-change restart?

Hi! This is a great question. What you're seeing is actually the expected behavior for Routine Load in StarRocks. Why this happened In StarRocks, Routine Load manages its own consumption offsets (stored internally in the FE) rather than relying on Kafka's server-side

__consumer_offsets

to determine where to start a new job. 1. New Job = New State: Even if you use the same job name or the same

group.id

, StarRocks treats a

CREATE ROUTINE LOAD

command as a brand-new entity with a fresh internal ID and no previous progress metadata. 2. Default Starting Offset: By default, if the

kafka_offsets

(or

property.kafka_default_offsets

) property is not specified in the

CREATE ROUTINE LOAD

statement, StarRocks initializes the job to start from the latest offsets (

OFFSET_END

). This causes it to skip any existing lag. 3. The

group.id

Role: While you can specify a

group.id

in the properties, StarRocks uses it primarily for identification and internal consumer logic. It does not automatically query Kafka to find that group's last committed offset when initializing a new job. The "Right Way" to preserve offsets If you need to perform a schema change and want to ensure no data is skipped, you have two main options: Option 1: Use

ALTER ROUTINE LOAD

(Recommended) Instead of dropping and recreating the job, you can usually

PAUSE

the job, perform your schema change on the table, and then use

ALTER ROUTINE LOAD

to update the job's mapping or properties. StarRocks supports altering

COLUMNS

and

jsonpaths

(since v2.x/v3.x), which allows you to adjust to the new schema without losing the internal consumption progress.

sql
-- 1. Pause the job
PAUSE ROUTINE LOAD FOR my_job;

-- 2. (Perform table schema changes)

-- 3. Alter the job if mappings changed (e.g., new jsonpaths or columns)
ALTER ROUTINE LOAD FOR my_job
COLUMNS (col1, col2, new_col, ...)
PROPERTIES (
    "jsonpaths" = "[\"$.k1\", \"$.k2\", \"$.k3\"]"
);

-- 4. Resume
RESUME ROUTINE LOAD FOR my_job;

Option 2: Manually specify offsets on Recreation If you must recreate the job (e.g., a major change that

ALTER

doesn't cover): 1. Before dropping the old job, run

SHOW ROUTINE LOAD FOR <job_name>\G

and copy the values from the

Progress

field. This field shows the exact offset StarRocks has successfully committed for each partition (e.g.,

{"0":"12345", "1":"67890"}

). 2. In your new

CREATE ROUTINE LOAD

statement, explicitly set those offsets:

sql
    FROM KAFKA (
        "kafka_broker_list" = "...",
        "kafka_topic" = "...",
        "kafka_partitions" = "0, 1",
        "kafka_offsets" = "12345, 67890" -- Use the values from the old Progress
    );

Summary: For your specific version (4.0.7), using

ALTER ROUTINE LOAD

is the cleanest way to stay continuous, as it keeps the job's internal state (and lag) intact while you update the loading logic. Hope this helps! Let me know if you have more questions. References • sql-reference/sql-statements/loading_unloading/routine_load/ALTER_ROUTINE_LOAD.md • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md