http://starrocks.io logo
Join Slack
Powered by
<@U0A71G31CDV> I have routine load that doesn’t se...
# questions-and-troubleshooting
d
@Rocky I have routine load that doesn’t seem to scale. I have 4 CN nodes on shared-data cluster. Kafka topic with 24 partitions. I’ve created ROUTINE LOAD with following properties:
Copy code
"max_batch_interval" = "30",
 "max_batch_rows" = "2000000",
 "desired_concurrent_number" = "24"
But the throughput is very low and cluster is barely utilized
r
Based on your configuration and the StarRocks architecture, your Routine Load is likely hitting a "parallelism cap" defined by the number of Compute Nodes (CNs) in your cluster. 1. The Parallelism Formula The actual number of concurrent tasks for a single Routine Load job is determined by the minimum of several parameters:
Copy code
text
Actual Concurrency = min(alive_be_number, partition_number, desired_concurrent_number, max_routine_load_task_concurrent_num)
In your case: * `alive_be_number`: 4 (your 4 CN nodes). * `partition_number`: 24. * `desired_concurrent_number`: 24 (set by you). * `max_routine_load_task_concurrent_num`: 5 (default FE global configuration). Result: Your job is likely running only 4 parallel tasks. This explains why your cluster is underutilized; each node is only running one load task, which may not be enough to saturate the CPU/IO, especially if the data processing logic is simple. 2. How to Scale the Throughput To increase the throughput beyond the number of nodes, you can use the following strategies: A. Split the Routine Load into multiple jobs Since a single Routine Load job is typically capped by the number of nodes (1 task per node), the most effective way to scale on a small cluster is to create multiple Routine Load jobs for the same Kafka topic, each responsible for a subset of partitions. * Job 1: Partitions 0-5 * Job 2: Partitions 6-11 * Job 3: Partitions 12-17 * Job 4: Partitions 18-23 This bypasses the 
alive_be_number
limit for a single job. Your 4 nodes can then run multiple tasks simultaneously (e.g., if each job runs 4 tasks, you'd have 16 total tasks running across 4 nodes). B. Check and Adjust Global FE Configurations Even if you split jobs, you might hit global limits. You may need to increase these FE dynamic parameters (using 
ADMIN SET FRONTEND CONFIG
): * `max_routine_load_task_concurrent_num`: Increase this from the default 5 to a higher value (e.g., 24). This allows a single job to attempt more tasks if the formula allows. * `max_routine_load_task_num_per_be`: (Default 16 in v3.1+) Ensure this is high enough to accommodate all tasks from all your routine load jobs on a single node. C. Adjust Batching Logic Your 
max_batch_interval
is set to 30s. Routine Load commits a transaction when it hits either the row limit or the time limit. * If your data arrives slowly, StarRocks will wait 30 seconds before committing. * If you want higher throughput and lower latency, consider decreasing 
max_batch_interval
(e.g., to 5s or 10s) and ensuring 
max_batch_rows
is not the bottleneck. 3. Verification Steps Run the following command to see what is happening under the hood:
Copy code
sql
SHOW ROUTINE LOAD FOR <job_name>\G
Look at: * 
State
: Ensure it is 
RUNNING
. * 
Statistic
: Check 
runningTasks
. If it says 4, you are hitting the node-count limit. * 
ReasonOfPause
: If it's paused, check the error message here. * 
Progress
: Verify that all 24 partitions are actually being consumed. Summary recommendation: Increase the global FE config 
max_routine_load_task_concurrent_num
to 24. If the task count is still capped at 4, split your Kafka topic consumption into 4 separate Routine Load jobs. References • faq/loading/Routine_load_faq.md • sql-reference/sql-statements/loading_unloading/BROKER_LOAD.md • loading/RoutineLoad.md
d
@Rocky I have created 6 routine loads with 4 topic partitions each, it’s still not enough throughput, cluster is using ~30% of CPU but consuming 160k records/s, 4 CN nodes, desired concurrency is 4
Copy code
"max_batch_interval"="15",
"max_batch_rows"="2000000",
"task_consume_second"="15",
"task_timeout_second"="60",
r
To increase throughput on your shared-data cluster while CPU usage remains low, you are likely hitting the transaction commit bottleneck rather than a processing bottleneck. In shared-data mode, every Routine Load task commit involves metadata operations and "publishing versions," which are more expensive than in shared-nothing mode. Since you are already using 6 jobs to bypass the node-concurrency cap, here is how to further scale: 1. Reduce Commit Frequency (Increase Batching) With your current settings (
task_consume_second=15
), your 24 tasks are committing roughly every 15 seconds. If you are processing 160k records/s total, each task commits ~100,000 rows. In shared-data, larger and less frequent commits are much more efficient. * Increase 
task_consume_second
and 
max_batch_interval
: Set these to 30 or 60. This will double or quadruple the amount of data processed per transaction, reducing the overhead of metadata management. * Increase 
max_batch_rows
: Your 2M setting is fine, as the time limit is likely triggering the commit first. 2. Enable Batch Publish Version (v3.2.3+) If you are on StarRocks version 3.2.3 or higher, there is a specific optimization for shared-data clusters that batches transaction commits across multiple jobs/tasks. This significantly improves throughput for real-time ingestion.
Copy code
sql
-- Run this on FE
ADMIN SET FRONTEND CONFIG ("lake_enable_batch_publish_version" = "true");
3. Check Global Thread Limits Even if you define many jobs, they may be queued if the BE thread pool is saturated. * 
routine_load_thread_pool_size
: This is a BE configuration (default is often 10). With 6 jobs and 4 tasks each, you have 24 tasks across 4 CNs, meaning ~6 tasks per CN. You are close to the limit. * Check if tasks are being delayed by looking for 
TOO MANY TASKS
in the FE logs or by running 
SHOW ROUTINE LOAD
and checking if 
runningTasks
matches your 
desired_concurrent_number
. 4. Optimize CN Configurations For high-throughput writing to object storage, adjust these BE configs (via 
update information_schema.be_configs
): * `flush_thread_num_per_store`: Increase to 
32
(helps with S3 write concurrency). * `number_tablet_writer_threads`: Increase to 
32
. 5. Diagnose via BE Logs To see exactly why a task finished (did it hit the row limit, time limit, or an error?), check 
be/log/be.INFO
on your CN nodes. Look for lines containing 
consumer group done
. * If 
left_bytes
is a large positive number, it means the task finished because of the time limit (
task_consume_second
). * If 
left_bytes
is negative, it hit the size limit (
max_routine_load_batch_size
). Recommended next step: Increase 
task_consume_second
to 60 and enable 
lake_enable_batch_publish_version
if your version supports it. This usually results in a significant throughput jump for shared-data clusters with low CPU utilization. References • knowledge/usage/load/shared_data_batch_ingestion.md • faq/loading/Routine_load_faq.md • faq/shared_data_faq.md • administration/management/FE_configuration.md • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md References • faq/loading/Routine_load_faq.md • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md
d
@Rocky batches are bigger, but now it operates in spikes. CPU would go up for a few seconds, record read would go to 600k/s for a few seconds, then it drops to zero for some time. I'm using range distributed table - is that an issue here?
r
The "spike and drop" behavior you are seeing is typical when Routine Load tasks are synchronized. Because a Routine Load task is effectively a sequence of discrete transactions (Consume → Stop → Commit → Schedule New Task), your throughput drops to zero during the Commit and Scheduling phases. When using a range-partitioned table in a shared-data cluster, this effect is often magnified by tablet contention. Here is how to diagnose and fix it: 1. Tablet Lock Contention (The "Range" Issue) If your table is partitioned by a range (e.g., 
date
) and all your Kafka data belongs to the same "current" partition, all 24 tasks are trying to write to the same set of tablets simultaneously. * The Problem: If that active partition has a low bucket count (e.g., 4 or 8 buckets), your 24 concurrent tasks will fight for write locks on those few tablets. This causes some tasks to stall, leading to uneven "spiky" CPU usage. * The Fix: Increase the number of buckets (Distributed By) for your partitions. For 24 concurrent tasks across 4 CNs, aim for at least 24–48 buckets per partition to ensure each task has its own tablet to write to without waiting for locks. 2. The "Stop-and-Wait" Gap Routine Load is not a continuous stream; it is a series of "Mini-Batches." * Consumption Phase: CPU and throughput spike. * Commit Phase: Consumption stops. The BE flushes data to S3 and the FE updates metadata. In shared-data, this metadata update ("Publish Version") can take a few seconds. * Scheduling Phase: The FE waits for the next 
max_batch_interval
or scheduling cycle to start the next task. To bridge the gap: * Reduce 
max_batch_interval
: If this is set to 
30
, the FE might wait longer than necessary to start the next round of tasks. Try reducing it to 
10
or 
5
. * Stagger your Jobs: Since you have 6 separate jobs, try to start them a few seconds apart. This prevents all 24 tasks from hitting the "Commit" phase at the exact same millisecond, which helps smooth out the throughput. 3. Shared-Data Metadata Bottleneck In shared-data mode, the "zero" periods are often caused by the FE being busy committing versions to the object store metadata. * Enable Batch Publish: As mentioned before, ensure you are on v3.2.3+ and run:
Copy code
sql
    ADMIN SET FRONTEND CONFIG ("lake_enable_batch_publish_version" = "true");
This allows the FE to commit multiple Routine Load transactions in one go, drastically reducing the "zero throughput" time. 4. Check for "Too Many Versions" Range partitioning can lead to a high number of tablets across the cluster. If you have thousands of partitions/tablets, the metadata overhead for every commit increases. * Check your version count: 
SHOW TABLET FROM <table_name>
and look at the 
Version
column. * If versions are climbing into the hundreds/thousands, your ingestion is too frequent. Increase 
task_consume_second
 (e.g., to 60s or 120s) to commit less often. Summary Checklist 1. Buckets: Ensure 
DISTRIBUTED BY HASH(...) BUCKETS X
has a high enough 
X
(at least 24) for the active range partition. 2. Scheduling: Set 
"max_batch_interval" = "5"
in your Routine Load properties to restart tasks faster. 3. Batching: Use 
lake_enable_batch_publish_version
to handle the shared-data metadata overhead. 4. Transaction Size: Keep 
task_consume_second
high (60s+) so that the "Commit" overhead is a small fraction of the total time. If you see 
Waiting for Publish
taking a long time in the FE logs, it is almost certainly a metadata/versioning bottleneck. References • knowledge/usage/load/shared_data_batch_ingestion.md • faq/loading/Routine_load_faq.md • faq/shared_data_faq.md
d
@Rocky from CN logs:
Copy code
I20260718 11:21:10.451927 281417120273472 data_consumer_group.cpp:146] consumer group done: 7b4f32dea31a4709-f64eb004c04df8a5. consume time(ms)=60000, received rows=735466, received bytes=547328917, eos: 0, left_time: 0, left_bytes: 3747638379, blocking get time(us): 1048740, blocking put time(us): 58056772
I20260718 11:21:12.605633 281418496136256 data_consumer_group.cpp:146] consumer group done: e7432a373d9a885b-064de083d4969692. consume time(ms)=60000, received rows=732125, received bytes=548919172, eos: 0, left_time: 0, left_bytes: 3746048124, blocking get time(us): 1042905, blocking put time(us): 58154218
I20260718 11:21:14.457513 281381479306304 data_consumer_group.cpp:146] consumer group done: 664a12340d63cf2b-f495c7dd30f9c8b9. consume time(ms)=60000, received rows=729047, received bytes=545333103, eos: 0, left_time: 0, left_bytes: 3749634193, blocking get time(us): 1043145, blocking put time(us): 58071450
I20260718 11:21:15.210614 281458526573632 data_consumer_group.cpp:146] consumer group done: dc452093682b1a9d-8ef56b81e618c989. consume time(ms)=60000, received rows=733279, received bytes=549018476, eos: 0, left_time: 0, left_bytes: 3745948820, blocking get time(us): 1050258, blocking put time(us): 58183769
I20260718 11:21:16.736968 281381651141696 data_consumer_group.cpp:146] consumer group done: 204ece6fb82056cc-cc1a5b55dad1baa4. consume time(ms)=60000, received rows=724722, received bytes=552218970, eos: 0, left_time: 0, left_bytes: 3742748326, blocking get time(us): 1070941, blocking put time(us): 58131728
I20260718 11:22:00.760508 281406524806208 data_consumer_group.cpp:146] consumer group done: 224c89ae1a5639d5-68e0927b43969dbb. consume time(ms)=60000, received rows=729110, received bytes=543177466, eos: 0, left_time: 0, left_bytes: 3751789830, blocking get time(us): 1040812, blocking put time(us): 175426485
I20260718 11:22:16.971774 281381479306304 data_consumer_group.cpp:146] consumer group done: 0947d4c3debcdf15-471dad99f4cea987. consume time(ms)=60000, received rows=726291, received bytes=554865114, eos: 0, left_time: 0, left_bytes: 3740102182, blocking get time(us): 1103494, blocking put time(us): 58088890
I20260718 11:22:17.560675 281458526573632 data_consumer_group.cpp:146] consumer group done: 074266b7f89c1c85-2e874e7f5a03c395. consume time(ms)=60007, received rows=742512, received bytes=548614557, eos: 0, left_time: -7, left_bytes: 3746352739, blocking get time(us): 47355, blocking put time(us): 59173269
I20260718 11:22:19.441304 281442692944960 data_consumer_group.cpp:146] consumer group done: b3423f022679806e-96af9794c5a6ea8d. consume time(ms)=60000, received rows=717909, received bytes=526966681, eos: 0, left_time: 0, left_bytes: 3768000615, blocking get time(us): 1065353, blocking put time(us): 58178365
I20260718 11:22:21.385117 281417172833344 data_consumer_group.cpp:146] consumer group done: d1466a46cca46ee7-062b9c74961db49f. consume time(ms)=60000, received rows=729141, received bytes=553002639, eos: 0, left_time: 0, left_bytes: 3741964657, blocking get time(us): 1044951, blocking put time(us): 58187422
I20260718 11:22:22.172765 281377976865856 data_consumer_group.cpp:146] consumer group done: a04c9ac4d0aef46c-63d925fe3c7deb95. consume time(ms)=60000, received rows=727476, received bytes=547191709, eos: 0, left_time: 0, left_bytes: 3747775587, blocking get time(us): 1040331, blocking put time(us): 58184886
I20260718 11:22:23.680096 281417120339008 data_consumer_group.cpp:146] consumer group done: 3d48688d15b839eb-1434804c7c30e998. consume time(ms)=60000, received rows=722615, received bytes=534083954, eos: 0, left_time: 0, left_bytes: 3760883342, blocking get time(us): 1062014, blocking put time(us): 57616694
r
The logs you provided contain the "smoking gun" for your performance issue. 1. Understanding the Bottleneck: 
blocking put time
 In your log entries: * 
consume time(ms)=60000
: The task ran for exactly 60 seconds (the default timeout). * 
blocking put time(us): 58,056,772
: This is ~58 seconds. This means for a 60-second task, the consumer was blocked for 58 seconds waiting to "hand off" data to the StarRocks writing process. It was only actually consuming data from Kafka for about 2 seconds. This confirms that the Sink (StarRocks Write Path) is the bottleneck, not Kafka or the network. The "spike and drop" happens because the internal buffer fills up instantly, the consumer stops (throughput drops to zero), and then it waits for the writer to slowly clear the buffer (throughput remains zero), before it can grab another tiny burst of data (the spike). 2. Why the Sink is Blocked Since you are using a Range Distributed Table on a Shared-Data Cluster, there are two primary culprits for this high "blocking put time": A. Tablet Lock Contention (Most Likely) With 24 tasks all writing to the same "current" range partition, they are all trying to update the same set of tablets. * If your partition has only a few buckets (e.g., 4 or 8), then 24 concurrent tasks are fighting over 4-8 tablet locks. * Only one task can write to a specific tablet at a time. The other 20+ tasks will sit in 
blocking put
state waiting for the lock. * The Fix: Increase the number of buckets for your active partitions. For 24 concurrent tasks, you should have at least 24–48 buckets in the partition to ensure each task can write to a unique tablet without waiting. B. Shared-Data "Publish Version" & Write Latency In shared-data mode, every "put" involves writing to the local cache and eventually flushing to object storage. If the writer threads are saturated or waiting for metadata updates, the 
blocking put time
will skyrocket. * The Fix (BE Config): Increase the writer concurrency on your 4 CN nodes to ensure they can handle multiple tasks in parallel.
Copy code
sql
    -- Run on FE to update all CNs
    UPDATE information_schema.be_configs SET value = "32" WHERE name = "flush_thread_num_per_store";
    UPDATE information_schema.be_configs SET value = "32" WHERE name = "number_tablet_writer_threads";
* The Fix (FE Config): Ensure 
lake_enable_batch_publish_version
is 
true
(requires v3.2.3+) to prevent the FE from becoming a bottleneck during the commit phase of these 24 tasks. 3. Why your configuration isn't being fully respected Your logs show 
consume time(ms)=60000
, but you mentioned setting 
task_consume_second="15"
. * This suggests your individual Job settings might be overridden by the global FE configuration, or the job wasn't updated correctly. * Check the global limit: 
ADMIN SHOW FRONTEND CONFIG LIKE 'routine_load_task_consume_second';
_ If you increase the throughput by fixing the tablet locks, you actually _want* this to be longer (like 60s) to keep the transactions large and efficient. Summary of Actions 1. Check Buckets: Run 
SHOW PARTITIONS FROM <table_name>
and check the 
Buckets
count for the active partition. If it's less than 24, recreate the table (or add new partitions) with a higher bucket count. 2. Increase BE Threads: Apply the 
flush_thread_num_per_store
and 
number_tablet_writer_threads
changes mentioned above. 3. Monitor Lock Contention: You can check for lock issues by looking at 
SHOW PROC '/transactions';
and looking for tasks that stay in 
PREPARED
state for a long time. References • faq/loading/Routine_load_faq.md • knowledge/usage/load/shared_data_batch_ingestion.md
d
@Rocky I'm struggling to achieve throughout for my topic. If I use Flink instead of direct routine load - will that be better for achieving higher throughput, avoiding concurrent transactions that step on each other toes?
r
Yes, switching to the StarRocks Flink Connector can significantly improve throughput and solve the "stepping on toes" issue, provided you leverage a specific feature called Merge Commit. Here is why Flink would be better for your current shared-data bottleneck: 1. The Core Solution: "Merge Commit" The primary reason you are seeing a 58s 
blocking put time
is the high number of concurrent transactions (6 jobs × 4 tasks = 24 transactions) hitting the same tablets and metadata service. * Routine Load: Each task is an independent transaction. 24 tasks = 24 separate "Publish Version" operations on the FE and 24 sets of tablet locks. * Flink with Merge Commit: Flink can have high parallelism (e.g., 24 slots) to consume Kafka quickly, but it merges the data from all those slots into a single StarRocks transaction on the server side. This drastically reduces metadata overhead and eliminates the "lock contention" between your parallel tasks. 2. When to Use Flink vs. Routine Load Feature | Routine Load | Flink Connector Transaction Management | 1 Task = 1 Transaction | Many Sink Tasks = 1 Transaction (via Merge Commit) Shared-Data Perf | High metadata overhead per task | Optimized for high-throughput shared-data writes Complexity | Simple (SQL only) | Requires a Flink cluster Scaling | Limited by BE/CN thread pools | Scales horizontally with Flink workers 3. How to Configure Flink for High Throughput To achieve the best results on a shared-data cluster, use the following Flink SQL properties (requires StarRocks v3.4.0+ and Flink Connector v1.2.9+):
Copy code
sql
CREATE TABLE sr_sink (
    ...
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_ip>:9030',
    'load-url' = '<fe_ip>:8030',
    -- Enable Merge Commit to bundle multiple Flink tasks into one SR transaction
    'sink.properties.enable_merge_commit' = 'true',
    -- How often StarRocks should commit the merged transaction (e.g., every 10s)
    'sink.properties.merge_commit_interval_ms' = '10000',
    -- How often Flink subtasks should flush data to StarRocks
    'sink.buffer-flush.interval-ms' = '5000',
    -- High parallelism for consuming Kafka, while keeping transactions low
    'sink.parallelism' = '24' 
);
4. Important Considerations * StarRocks Version: Server-side Merge Commit is best supported in v3.4.0 or later. If you are on an older version, Flink will still help by allowing larger, more controlled manual batching, but you won't get the server-side merging benefit. * Memory: Ensure your CN nodes have enough memory for the 
memtable
because Flink will be pushing much larger batches. * Bucketing: Even with Flink, ensure your table partitions have enough buckets (e.g., 24-32) so that the single large merged transaction can write to multiple tablets in parallel. Verdict: If you have the infrastructure for Flink, it is the recommended way to reach very high ingestion rates (500k+ records/s) on shared-data clusters because it decouples consumption parallelism from transaction count. References • loading/Flink-connector-starrocks.md • knowledge/usage/load/shared_data_batch_ingestion.md • faq/loading/Flink_connector_faq.md • unloading/Flink_connector.md • administration/management/monitoring/metrics.md References • loading/Flink-connector-starrocks.md • faq/loading/Flink_connector_faq.md
d
@Rocky why routine load doesn't have merge commit analogue? Wouldn't that solve concurrency issue? Are there plans to add support for that?
r
Routine Load doesn't have a "merge commit" feature today primarily because of its internal architecture, but there are technical reasons for this and clear alternatives for high-throughput needs. Why Routine Load lacks Merge Commit 1. Internal Scheduling vs. External Push: Merge Commit was designed for the Stream Load API, which is a "push" model where external clients (like Flink or a Python script) send data to StarRocks. The server can easily buffer and merge these incoming HTTP requests. Routine Load, however, is an FE-managed "pull" model. The FE explicitly schedules individual tasks to specific CN/BE nodes. Each task is defined by a specific Kafka offset range and is treated as a standalone, atomic transaction. 2. Transaction Isolation: Because Routine Load guarantees Exactly-Once semantics by committing specific Kafka offsets alongside the data, merging multiple tasks (which might cover different partitions or jobs) into one transaction would significantly complicate the offset management and error-recovery logic within the FE. Are there plans to add it? There is no currently documented 
enable_merge_commit
parameter for Routine Load. However, StarRocks is moving toward solving the same "concurrency vs. throughput" problem in two ways: * Ingestion Optimization (v3.5+): Starting in version 3.5, StarRocks introduced "Batch Ingestion Optimization" and "Load Spill." Instead of merging transactions, these features focus on making individual transactions much more efficient by merging data blocks during the ingestion phase and spilling to disk to prevent the "small file" and memory pressure issues that usually trigger the need for merge commit. * The Connector Strategy: The official recommendation for high-concurrency scenarios is shifting toward the StarRocks Kafka Connector (for Kafka Connect) or Flink Connector. These external connectors do support Merge Commit (as of v3.4.0). They allow you to have 24+ workers consuming Kafka but only one transaction hitting the StarRocks metadata layer every few seconds. Summary If you are hitting a wall with Routine Load concurrency: 1. Upgrade to v3.5+: To benefit from the new internal batching and spill optimizations which reduce the metadata/small-file overhead. 2. Switch to the Kafka Connector: This is the "Routine Load 2.0" approach. It runs in Kafka Connect and supports the 
enable_merge_commit
property, effectively solving the "stepping on toes" issue by bundling multiple Kafka partition reads into a single StarRocks transaction. Verdict: While Routine Load remains the easiest to set up, it is architecturally "pinned" to a 1-task-1-transaction model. For your 500k+ record/s requirement, the Kafka Connector or Flink Connector with Merge Commit enabled is the intended architectural path forward. References • loading/StreamLoad.md • loading/Flink-connector-starrocks.md • loading/Loading_intro.md • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md • loading/RoutineLoad.md • administration/management/FE_configuration.md • loading/Kafka-connector-starrocks.md References • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md • administration/management/BE_configuration.md
d
@Rocky For 1Tb per day of streaming data with 150k records/s with Kafka Connect, what are good settings for records / bytes for efficient loading?
r
For your scale of 1 TB per day (~12 MB/s) with a high record count of 150,000 records/s, the goal is to create fewer, larger batches. In a shared-data (lake) architecture, small frequent commits are the primary cause of high latency and "blocking put" times because they overwhelm the metadata service and create too many small files in object storage. Here are the recommended settings for the StarRocks Kafka Connector to achieve efficient loading at this volume: 1. Flush Trigger Settings You want to target a batch size of 200 MB to 500 MB or a time interval of 10–30 seconds. * 
bufferflush.maxbytes
: Set to 
314572800
(300 MB). _ _Why:* At 12 MB/s, it will take about 25 seconds to fill this buffer. This creates a healthy file size for object storage. * 
bufferflush.intervalms
: Set to 
30000
(30 seconds). _ _Why:* This acts as a safety timeout. If traffic drops, it ensures data is still loaded at least every 30 seconds. * 
tasks.max
: Set to 24 (to match your Kafka partitions and previous parallelism). _ _Note:* With 24 tasks, each task is only handling ~0.5 MB/s. If they flush independently, you get 5 MB files, which is too small. You must use Merge Commit (below) to fix this. 2. The "Efficiency Secret": Merge Commit This is the most important setting for high-throughput shared-data clusters. It allows all 24 tasks to send their small buffers to StarRocks, which then merges them into a single transaction before committing to the FE. Add these to your connector configuration: * 
sink.properties.enable_merge_commit
: 
true
* 
sink.properties.merge_commit_interval_ms
: 
10000
(10 seconds) _ _Result:* StarRocks will bundle all data from all tasks into one "Publish Version" every 10 seconds, drastically reducing metadata overhead. 3. Kafka Connect Worker Setting Kafka Connect has its own internal offset commit interval that can override the connector's settings. * 
<http://offset.flush.interval.ms|offset.flush.interval.ms>
: Set this to at least 
30000
in your Kafka Connect worker configuration. If this is left at the default (often 60s), it's fine, but if it's too low (e.g., 1s), it will force StarRocks to commit every second regardless of your other settings. Summary Configuration Snippet
Copy code
properties
# Capacity & Batching
tasks.max=24
bufferflush.maxbytes=314572800
bufferflush.intervalms=30000

# Stream Load / Merge Commit (Requires SR v3.2.4+)
sink.properties.enable_merge_commit=true
sink.properties.merge_commit_interval_ms=10000

# Data Layout
# Ensure your table has at least 24-48 buckets
Why these numbers? * 150k rec/s at 80 bytes/rec is high-frequency. If you commit every 1 second, you are asking the StarRocks FE to manage 86,400 transactions per day. * By moving to a 10-30s commit window via Merge Commit, you reduce that to ~3,000–8,000 transactions per day. This will eliminate the 
blocking put
bottleneck and allow the CN nodes to ingest data at wire speed. References • loading/Kafka-connector-starrocks.md • loading/Flink-connector-starrocks.md • loading/Spark-connector-starrocks.md • loading/StreamLoad.md • administration/management/FE_configuration.md • loading/RoutineLoad.md References • loading/Kafka-connector-starrocks.md • loading/Flink-connector-starrocks.md • loading/Spark-connector-starrocks.md
d
@Rocky When using Kafka Connect - does data go through FE nodes, so we need to consider it’s resources not just CN?
r
When using the StarRocks Kafka Connector, the actual data payload does not pass through the FE nodes. Here is how the data flow works and what you need to consider for resources: 1. The Data Flow The Kafka Connector uses the Stream Load interface under the hood. The process looks like this: 1. Initial Handshake (FE): The Kafka Connector sends an HTTP request to the FE (
starrocks.http.url
). 2. Redirect (FE to Connector): The FE receives the request, identifies which CN/BE nodes are healthy and suitable to handle the load, and sends an HTTP 307 Redirect back to the Kafka Connector. 3. Data Transmission (Connector to CN): The Kafka Connector follows the redirect and sends the entire data payload directly to the CN/BE node. 4. Transaction Commit (CN to FE): Once the CN has written the data, it communicates with the FE to commit the transaction and update metadata. 2. Resource Considerations Because the data bypasses the FE, you do not need to scale FE CPU or Network Bandwidth to handle the 1 TB/day data volume. However, the FE still plays a critical role: * FE Resources (Metadata & Transactions): Even though data doesn't flow through it, every batch you send creates a transaction. If you have many small batches (e.g., thousands of transactions per minute), the FE's JVM heap and Edit Log (metadata storage) can become a bottleneck. This is why using Merge Commit is so important—it reduces the number of metadata operations the FE must handle. * CN Resources (The Heavy Lifters): The CN/BE nodes handle 100% of the data parsing, compression, and writing to object storage (S3/GCS). For 1 TB/day and 150k rec/s, you should monitor the CPU and Memory on your CN nodes, especially the 
memtable
memory usage during heavy ingestion. * Kafka Connect Resources: The Kafka Connect workers themselves will need enough CPU and Memory to handle the serialization/deserialization of 150k records/s. If you use complex transformations (SMTs) in Kafka Connect, this is where you'll see the most overhead. Summary * FE: Only handles the "handshake" and "metadata." It does not touch the 1 TB of data. * CN: Handles the actual 1 TB of data. Scale these nodes for ingestion throughput. * Connector: Handles the Kafka consumption. Scale your Kafka Connect cluster to handle the 150k rec/s processing. Verdict: You do not need high-spec FE nodes for data volume, but you do need healthy FEs (with enough RAM) to manage the transactional metadata, especially if you aren't using Merge Commit to batch your writes into larger batches. References • loading/Kafka-connector-starrocks.md • loading/StreamLoad.md • loading/RoutineLoad.md • loading/Loading_intro.md • administration/management/FE_configuration.md References • loading/Kafka-connector-starrocks.md • loading/RoutineLoad.md
d
@Rocky Does Kafka Connect not support merge commit because it applies labels? 
StarRocksSinkTask
hardcodes 
.enableTransaction()
— label-based transactional Stream Load — and Merge Commit only applies to label-less loads?
r
You are correct in your technical assessment. There is a fundamental architectural conflict between the current StarRocks Kafka Connector and the Merge Commit feature. 1. The Label Conflict Merge Commit is designed to take multiple independent, "homogeneous" Stream Load requests and bundle them into a single server-managed transaction. To do this, the server must generate its own label for the merged transaction. As noted in the StarRocks documentation for Stream Load:
"The server automatically generates labels for transactions. They will be ignored if specified [by the client]."
The Kafka Connector, however, uses label-based loading for idempotency. It generates a unique label for each batch (often incorporating the Kafka offsets) so that if a network error occurs, it can retry the same batch with the same label to prevent double-loading. If Merge Commit were active, the server would ignore the Connector's label, making it impossible for the Connector to verify if its specific data was successfully committed or if a retry is safe. 2. Transactional Interface vs. Merge Commit You noticed that 
StarRocksSinkTask
uses 
.enableTransaction()
. This means the connector uses the Stream Load Transaction Interface (
/api/transaction/begin
, 
prepare
, 
commit
), which is a Two-Phase Commit (2PC) mechanism. * Merge Commit is a feature of the standard Stream Load (
/api/_stream_load
) where the server manages the transaction lifecycle. * Transactional Interface is for client-managed transactions. Because the Kafka Connector is designed to manage the transaction lifecycle itself (matching the commit of the transaction to the commit of Kafka offsets), it cannot hand that control over to the server’s "Merge Commit" window. 3. Plans and Alternatives While there is no public roadmap to add "server-side merge" to the existing label-based Transactional Interface (since they are philosophically opposite), there are two ways this is being addressed: * Flink Connector Approach: The StarRocks Flink Connector already supports Merge Commit because it offers a "V1" (Standard) mode alongside "V2" (Transactional). If you enable Merge Commit in Flink, it falls back to At-Least-Once semantics because it loses the per-batch label safety. * Internal Batching (v3.5+): Recent StarRocks versions have introduced Load Spill and internal batching optimizations. Instead of merging separate transactions at the metadata level (Merge Commit), the system is becoming more efficient at handling the "small file" problem internally by spilling data to disk and merging files during the ingestion process itself. * Kafka Connector Updates: There is ongoing work to make the Kafka Connector more performant, but if your goal is strictly Merge Commit, the system currently requires a non-transactional (label-less) ingestion path which the current Kafka Connector (focused on consistency) doesn't prioritize. Summary: You cannot use Merge Commit with the Kafka Connector today because the Connector's requirement for idempotent labels and client-side transaction management contradicts the way Merge Commit operates on the server. If you need that scale, the Flink Connector (with 
sink.properties.enable_merge_commit=true
) is currently the primary path for merge-style high-concurrency ingestion. References • loading/Kafka-connector-starrocks.md • loading/Flink-connector-starrocks.md • sql-reference/sql-statements/loading_unloading/STREAM_LOAD.md • loading/Stream_Load_transaction_interface.md References • loading/Kafka-connector-starrocks.md
3 Views
Open in Slack
PreviousNext
Powered by