@Rocky No StarRocks Flink connector for Flink 2.x The newest connector, 1.2.x, supports Flink up to 1.20. We see open work for Flink 2.x support in the connector repo (for example PR #503). Only starrocks-stream-load-sdk 1.0 is published on Maven Central. It has no Merge Commit client, which exists only on the connector’s main branch. So on Flink 2.0.2 we would have to build our own Sink V2 on top of SDK 1.0: one Stream Load per table, at-least-once, idempotent upserts. Questions: Is there a timeline for a connector release that supports Flink 2.x? Will a new starrocks-stream-load-sdk, with the Merge Commit client, be published separately? Until then, is building a custom sink on SDK 1.0 the recommended route? Are there known limits, such as label handling or redirects, that we should plan for?

Currently, there is no official release of the StarRocks Flink connector for Flink 2.x. Support for Flink 2.0 is actively being developed under Issue #450 and PR #503, which introduces a multi-module layout to accommodate the major API changes in Flink 2.x. 1. Timeline for Flink 2.x Support There is no official release date yet. The work in PR #503 is ongoing. You can monitor the progress on GitHub or trial the experimental code by building the connector from that specific feature branch. 2.

starrocks-stream-load-sdk

Publication The

starrocks-stream-load-sdk

version 1.0 on Maven Central is significantly behind the version used in the current 1.2.x connectors. * Merge Commit Support: This feature (available since StarRocks 3.4) and the corresponding client code are part of the newer SDK logic bundled within the Flink connector repository (e.g., version 1.2.x). * Recommendation: Instead of using the 1.0 SDK from Maven Central, it is recommended to build the SDK from the latest

main

or

v1.2.x

branch of the starrocks-connector-for-apache-flink repository. This will give you access to the

MergeCommitHttpLoader

and other recent improvements. 3. Custom Sink on SDK 1.0: Risks and Recommendations Building a custom sink on the 1.0 SDK is possible but carries several technical hurdles: * Merge Commit: The 1.0 SDK does not support the Merge Commit API. If your workload involves many small, high-frequency writes (common in Flink), you will miss out on the performance benefits of merging multiple Stream Load requests into a single transaction. * Label Handling & Idempotency: You must manually manage Stream Load labels to ensure "at-least-once" or "exactly-once" semantics. If a Flink task fails and restarts, your custom sink must be able to generate the same label for the same data batch to avoid duplicates in StarRocks. * HTTP Redirects (307): StarRocks FEs redirect Stream Load requests to BEs using

307 Temporary Redirect

. The client must correctly retain authentication headers (Basic Auth) during this redirect. While the older SDK handles basic redirects, newer versions have optimized failover and health-checking logic for multiple FE/BE nodes. * V2 Sink Architecture: Flink 2.x relies heavily on the Sink V2 (Unified Sink) API. The existing StarRocks connector (1.2.x) already uses a V2-style architecture for its internal

StarRocksSink

, so porting that logic (from the

starrocks-stream-load-sdk

module in the 1.2.x branch) is a better starting point than starting from the 1.0 SDK. Suggested Route: If you cannot wait for the official release, the most robust path is to fork the connector repo, use the code from PR #503 as a base, and compile the

starrocks-stream-load-sdk

module locally to use as a dependency for your Flink 2.x project. References • unloading/Flink_connector.md • loading/Flink-connector-starrocks.md • release_notes/release-2.2.md • faq/loading/Flink_connector_faq.md

@Rocky Thanks for the connector answer. Here is more context on our case, and a few questions on the StarRocks side. Setup • One Kafka topic, about 2.6k msgs/s on average and 4.3k/s at peak. • Today 6 Routine Loads read that same topic. Each filters one event type into its own PRIMARY KEY table, and some do partial updates with

merge_condition

. • We want one Flink 2.0.2 job instead. It would buffer rows per table and send one Stream Load per table every 1–2 s, with one writer per table. Delivery is at-least-once. • Every table takes a full-row upsert with

merge_condition = src_offset

, where

src_offset

is the Kafka offset. Replays are therefore no-ops, and we don’t rely on labels for idempotency. Questions 1. Can Routine Load do this itself? ◦ Can one Routine Load write to several tables, or filter on a field that is not a column of the target table? ◦ If not, is 6 Routine Loads on one topic a pattern you would advise against? 2.

merge_condition

as idempotency ◦ Is this safe as the only protection against replays? ◦ When the same primary key appears twice in one load with different

src_offset

values, is the row with the higher value guaranteed to win? ◦ Is the same true across two loads that commit concurrently? 3. Load frequency on PRIMARY KEY tables ◦ About 30–60 loads per minute per table, on hourly partitions, 4 buckets, 3 replicas. Is that within your guidance? ◦ Should we tune

update_compaction_per_tablet_min_interval_seconds

, or

load_initial_open_partition_number

so a load opens only the newest partitions? 4. Merge Commit ◦ Is it planned to leave Beta in 4.0.x? ◦ Does it support

merge_condition

on PRIMARY KEY tables? We saw fix #63817. ◦ In sync mode, a publish timeout comes back as

Fail

. Should a client treat that as retry-safe? 5. Multi-table transactions (

transaction_type: multi

) ◦ Is the silent-loss issue fixed in 4.0.17 relevant to 4.0.12? ◦ Would you recommend 4.0.17 or later for any Stream Load-heavy workload? 6. SDK from source ◦ Is the

starrocks-stream-load-sdk

module free of any Flink dependency at tag v1.2.14 or later? ◦ Will it be published to Maven again?

To address your specific technical requirements for Flink 2.0.2 and StarRocks 4.0.x, here is the breakdown based on the current documentation and system behavior: 1. Routine Load Capabilities * Multi-table: A single Routine Load job can only write to one target table. It cannot dispatch data to multiple tables. * Filtering on non-column fields: Yes, this is supported. You can define a column in the

COLUMNS

clause that does not exist in the target table (a "derived" or "temporary" column) and use it in your

WHERE

clause. _ _Example:*

COLUMNS (id, event_type, data), WHERE event_type = 'order'

(where

event_type

is not a column in the table). * 6 Routine Loads on one topic: This is a supported pattern but is inefficient. Each job acts as a separate Kafka consumer group, meaning the same data is pulled from Kafka six times, increasing network and Kafka IO. Moving to Flink is the correct architectural choice for this scale. 2.

merge_condition

as Idempotency * Safety: Using

merge_condition = src_offset

is a safe and recommended way to handle idempotency for Primary Key tables. * Within a single load: StarRocks sorts the rows in a single batch by Primary Key and then by the

merge_condition

value. The row with the highest value is guaranteed to be the final version for that batch. * Across concurrent/sequential loads: StarRocks uses the

merge_condition

column as a sequence version. Even if loads commit concurrently or out of order, the storage engine ensures that a record with a lower

src_offset

will never overwrite a record with a higher

src_offset

that has already been applied. 3. Load Frequency on Primary Key Tables * Guidance: 30–60 loads per minute (every 1–2s) is high but manageable for Primary Key tables due to their persistent index. However, this frequency will generate many small versions, putting pressure on compaction. * Tuning: * `update_compaction_per_tablet_min_interval_seconds`: The default is 120s. If you see the "version count" of your tablets increasing significantly (check

SHOW PROC "/compactions"

), you may need to decrease this to trigger compaction more frequently. * `load_initial_open_partition_number`: This is only necessary if a single load batch writes to many different partitions (e.g., >10). If your Flink job buffers data such that most rows go to the "current" hourly partition, the default is fine. * Better Approach: If possible, try to increase your Flink buffer time to 5–10 seconds to reduce the total number of commit RPCs. 4. Merge Commit * Status: Merge Commit is no longer strictly "Beta" in the 4.0.x line and is considered stable for production use in high-concurrency stream loading. *

merge_condition

Support: Yes, recent fixes (including #63817) ensure that

merge_condition

works correctly with Merge Commit. * Publish Timeout (

Fail

): If a client receives a

publish timeout

, the transaction has already committed. The data is safe but not yet visible. * Retry Logic: A client should treat this as "Success (Pending)". If you retry with the same label, StarRocks will return

Label Already Exists

or

Transaction Already Committed

. You do not need to resend the data. 5. Multi-table Transactions * 4.0.17 vs 4.0.12: A critical fix for silent data loss in multi-table transaction dispatching was introduced in 4.0.17. * Recommendation: If you are using

transaction_type: multi

, it is highly recommended to use version 4.0.17 or later. Version 4.0.12 is potentially vulnerable to issues where certain table channels fail to dispatch correctly under high load. 6. SDK from Source

@Rocky Questions: 1. Can a single Routine Load write a different column set per row, depending on a field in the message, for example with

partial_update_mode

or conditional expressions in COLUMNS? 2. Can

merge_condition

differ per row within one load, or is it always one column for the whole load? 3. If neither is possible, is there another way to avoid reading the same topic four times? For example, can several Routine Loads share one consumer, or is there a multi-table Routine Load?

1. Different Column Set per Row No, a single Routine Load job cannot use a different column set per row. * Static Mapping: The

COLUMNS

clause in a Routine Load job defines a fixed schema mapping for all rows consumed by that job. * Conditional Values: While you can use

CASE

expressions within the

COLUMNS

clause to compute different values for a column based on a field in the message, the set of target columns remains the same for every row in the batch. * Partial Updates: If you enable

partial_update

, the specific columns being updated must be pre-defined in the

COLUMNS

list at the job level. You cannot dynamically switch which columns are being updated on a row-by-row basis within the same Routine Load job. 2.

merge_condition

per Row No,

merge_condition

is a job-level property and cannot differ per row. * Batch Constraint: StarRocks explicitly requires that you cannot specify different columns as update conditions for the same batch of data. * Column Name only: The property accepts a single column name as the version/sequence provider for the entire load. * Expression Workaround: You can point

merge_condition

to a derived column created in your

COLUMNS

clause (e.g.,

COLUMNS (..., my_version = CASE ... END)

and set

merge_condition = my_version

). However, even in this case, the logic used to calculate

my_version

must be identical for all rows in the load. 3. Avoiding Redundant Kafka Reads If you want to avoid reading the same topic multiple times (which happens with multiple Routine Loads) but are not yet ready for Flink 2.x, consider these options: * StarRocks Kafka Connector (Kafka Connect): This is the official alternative to Routine Load for complex dispatching. It runs as a plugin for Kafka Connect and supports the

starrocks.topic2table.map

property. This allows a single sink task to dispatch data from one or more topics to multiple different StarRocks tables. * Pipe (S3/Files only): The StarRocks

PIPE

feature is currently designed for cloud storage (S3/GCS/HDFS) and does not yet support Kafka. * Native Routine Load Limitations: Native Routine Load does not support shared consumers across jobs. Each job is an independent entity with its own consumer group and internal task management. Summary Recommendation: Since your volume is ~4.3k msgs/s, moving to a single Flink 2.0.2 job (building the SDK from source as discussed previously) is indeed the most efficient architecture. It solves the redundant read issue and allows you to implement the multi-table dispatching and

merge_condition

logic in code. References • sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md • loading/Load_to_Primary_Key_tables.md • loading/Kafka-connector-starrocks.md • loading/RoutineLoad.md References • loading/Kafka-connector-starrocks.md

@Tarundeep Singh https://starrocks.slack.com/archives/C02FAD0JSSD/p1793262934922399 there is already work in place for a long time now, but we are blocked in review and merging