http://starrocks.io logo
Join Slack
Powered by
Hi everyone! I’m on a PoC, in which I’m implementi...
# questions-and-troubleshooting
m
Hi everyone! I’m on a PoC, in which I’m implementing a data replication using Starrocks, the idea is to:
Mysql -> Debezium -> Kafka -> KafkaConnect -> Starrocks
I have a working solution in which I have enabled column mapping, but as soon as I enable the 
unwrap
+ 
addfield
, sthe kafka connect captures and maps columns automatically, I start getting this issue:
Copy code
Error: NULL value in non-nullable column 'id'. Row: [NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NUL
	at com.starrocks.data.load.stream.v2.StreamLoadManagerV2.AssertNotException(StreamLoadManagerV2.java:495)
	at com.starrocks.data.load.stream.v2.StreamLoadManagerV2.finishFlush(StreamLoadManagerV2.java:441)
	at com.starrocks.data.load.stream.v2.StreamLoadManagerV2.flush(StreamLoadManagerV2.java:383)
	at com.starrocks.connector.kafka.StarRocksSinkTask.preCommit(StarRocksSinkTask.java:359)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:430)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:400)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:231)
	... 9 more
Caused by: com.starrocks.data.load.stream.exception.StreamLoadFailException: Transaction prepare failed, db: orders, table: order_charges, label: -4f9e219f-d34c-4128-a954-b37460cce68e,
responseBody: {
    "Status": "INTERNAL_ERROR",
    "Message": "too many filtered rows",
    "ErrorURL": "<http://91.107.254.91:8040/api/_load_error_log?file=error_log_3b4ce9f279802e19_b3233a01a6cfc69a>"
}
errorLog: Error: NULL value in non-nullable column 'id'. Row: [NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, 0]
It’s clearly unable to map columns with the json object attributes. And help is highly appreciated to get it work. I have hundreds of tables, and mapping every single table manually and one by one is not really maintainable.
Debezium event on Kafka:
Copy code
{
  "before": {
    "id": 1234,
    "wallet_deducted_amount_cents": 0,
    "original_items_price_cents": 6180,
    "items_price_cents": 5780,
    "delivery_fee_cents": 900,
    "discount_cents": 900,
    "basket_discount_cents": 0,
    "items_discount_cents": 0,
    "delivery_discount_cents": 900,
    "earned_cashback_cents": 0,
    "vat_rate": 19,
    "vat_cents": 753,
    "total_price_cents": 5780,
    "payable_amount_cents": 5780,
    "paid_amount_cents": 0,
    "refunded_amount_cents": 0,
    "currency": "USD",
    "created_at": "2026-02-19T09:49:05Z",
    "updated_at": "2026-02-19T09:49:05Z"
  },
  "after": {
    "id": 1234,
    "wallet_deducted_amount_cents": 0,
    "original_items_price_cents": 6180,
    "items_price_cents": 5780,
    "delivery_fee_cents": 900,
    "discount_cents": 900,
    "basket_discount_cents": 0,
    "items_discount_cents": 0,
    "delivery_discount_cents": 900,
    "earned_cashback_cents": 0,
    "vat_rate": 19,
    "vat_cents": 753,
    "total_price_cents": 5780,
    "payable_amount_cents": 5780,
    "paid_amount_cents": 5780,
    "refunded_amount_cents": 0,
    "currency": "USD",
    "created_at": "2026-02-19T09:49:05Z",
    "updated_at": "2026-02-19T09:50:11Z"
  },
  "source": {
    "version": "3.4.0.Final",
    "connector": "mysql",
    "name": "dbz.orders",
    "ts_ms": 1771494611213,
    "snapshot": "false",
    "db": "orders",
    "sequence": null,
    "ts_us": 1771494611213315,
    "ts_ns": 1771494611213315000,
    "table": "order_charges",
    "server_id": 1866973752,
    "gtid": "4cc4899c-b939-3272-8c32-41c597973732:5381929053",
    "file": "mysql-bin-changelog.070110",
    "pos": 29251317,
    "row": 0,
    "thread": 223699,
    "query": null
  },
  "transaction": null,
  "op": "u",
  "ts_ms": 1771494611220,
  "ts_us": 1771494611220024,
  "ts_ns": 1771494611220024300
}
My table on SR:
Copy code
CREATE TABLE `order_charges` (
	`id` bigint(20) NOT NULL COMMENT "",
	`wallet_deducted_amount_cents` int(11) NOT NULL DEFAULT "0" COMMENT "",
	`original_items_price_cents` int(11) NOT NULL DEFAULT "0" COMMENT "",
	`items_price_cents` int(11) NOT NULL DEFAULT "0" COMMENT "",
	`delivery_fee_cents` int(11) NOT NULL DEFAULT "0" COMMENT "",
	`discount_cents` int(11) NOT NULL DEFAULT "0" COMMENT "",
	`basket_discount_cents` int(11) NULL COMMENT "",
	`items_discount_cents` int(11) NULL COMMENT "",
	`delivery_discount_cents` int(11) NULL COMMENT "",
	`earned_cashback_cents` int(11) NULL COMMENT "",
	`vat_rate` int(11) NOT NULL COMMENT "",
	`vat_cents` int(11) NOT NULL DEFAULT "0" COMMENT "",
	`total_price_cents` int(11) NOT NULL DEFAULT "0" COMMENT "",
	`payable_amount_cents` int(11) NULL COMMENT "",
	`paid_amount_cents` int(11) NULL COMMENT "",
	`refunded_amount_cents` int(11) NULL COMMENT "",
	`currency` varchar(255) NOT NULL COMMENT "",
	`created_at` datetime NOT NULL COMMENT "",
	`updated_at` datetime NOT NULL COMMENT ""
) ENGINE = OLAP PRIMARY KEY(`id`) COMMENT "OLAP" DISTRIBUTED BY HASH(`id`) BUCKETS 8 PROPERTIES (
	"compression" = "LZ4",
	"datacache.enable" = "true",
	"enable_async_write_back" = "false",
	"enable_persistent_index" = "true",
	"persistent_index_type" = "CLOUD_NATIVE",
	"replication_num" = "1",
	"storage_volume" = "builtin_storage_volume"
);
Docker file:
Copy code
FROM <http://quay.io/strimzi/kafka:0.47.0-kafka-3.9.0|quay.io/strimzi/kafka:0.47.0-kafka-3.9.0>
USER root

# StarRocks Kafka Connector
RUN mkdir -p /opt/kafka/plugins/starrocks-kafka-connector && \
    cd /opt/kafka/plugins/starrocks-kafka-connector && \
    curl -sLO <https://github.com/StarRocks/starrocks-connector-for-kafka/releases/download/v1.0.6/starrocks-connector-for-kafka-1.0.6-with-dependencies.jar>

# Debezium MySQL Connector (includes transform classes like ExtractNewRecordState)
RUN mkdir -p /opt/kafka/plugins/debezium-mysql-connector && \
    cd /opt/kafka/plugins/debezium-mysql-connector && \
    curl -sL <https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/3.0.5.Final/debezium-connector-mysql-3.0.5.Final-plugin.tar.gz> | tar xz --strip-components=1

USER 1001
Kafka connect configs:
Copy code
class: com.starrocks.connector.kafka.StarRocksSinkConnector
tasksMax: 1
config:
  topics: dbz.orders.orders.order_charges
  starrocks.http.url: 127.0.0.1:8030
  starrocks.database.name: orders
  starrocks.topic2table.map: dbz.orders.orders.order_charges:order_charges
  starrocks.username: XYZ
  starrocks.password: qwerty123

  # Flush policy
  bufferflush.maxbytes: "94371840"
  bufferflush.intervalms: "5000"

  # Transforms: add operation field, then unwrap Debezium envelope
  transforms: addfield,unwrap
  transforms.addfield.type: com.starrocks.connector.kafka.transforms.AddOpFieldForDebeziumRecord
  transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
  transforms.unwrap.drop.tombstones: true
  transforms.unwrap.delete.handling.mode: drop

  sink.properties.format: json
  sink.properties.strip_outer_array: true

  # Explicit mappings - after unwrap, fields are at root level
  # sink.properties.columns: "id,wallet_deducted_amount_cents,original_items_price_cents,items_price_cents,delivery_fee_cents,discount_cents,basket_discount_cents,items_discount_cents,delivery_discount_cents,earned_cashback_cents,vat_rate,vat_cents,total_price_cents,payable_amount_cents,paid_amount_cents,refunded_amount_cents,currency,created_at,updated_at"
  # sink.properties.jsonpaths: '["$.id","$.wallet_deducted_amount_cents","$.original_items_price_cents","$.items_price_cents","$.delivery_fee_cents","$.discount_cents","$.basket_discount_cents","$.items_discount_cents","$.delivery_discount_cents","$.earned_cashback_cents","$.vat_rate","$.vat_cents","$.total_price_cents","$.payable_amount_cents","$.paid_amount_cents","$.refunded_amount_cents","$.currency","$.created_at","$.updated_at"]'

  key.converter: org.apache.kafka.connect.json.JsonConverter
  key.converter.schemas.enable: "false"
  value.converter: org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable: "false"
Found the issue and fixed it on my setup! There 2 major problems: 1. The version of the debezium used in the repo is too old 2. for the unwrap mode to work, we have to have the schema mode enabled on both end: a. Debezium b. Starrocks kafka connect > • Issue 1: 
AddOpFieldForDebeziumRecord
silently becomes a NO-OP > > Your pipeline runs with 
value.converter.schemas.enable: false
on both the source and sink KafkaConnect workers. This means: > Debezium source serializes CDC events as plain JSON (no schema envelope) onto Kafka > 1. The sink KafkaConnect worker deserializes them into 
Map<String, Object>
 (not 
Struct
) > But look at AddOpFieldForDebeziumRecord.java:69: >
Copy code
if (!(record.value() instanceof Struct)) {
>     return record;  // ← SILENTLY RETURNS UNCHANGED for Map-based records!
> }
> When 
schemas.enable: false
, 
record.value()
is a 
Map
, not a 
Struct
. The transform does nothing and returns the original Debezium envelope untouched. No 
__op
field is added, and the record structure isn’t prepared for unwrap. > • Issue 2: Major Debezium version mismatch > > ComponentVersionNotesStarRocks connector (source pom.xml)built against *Debezium 1.8.1.Final*3+ years oldDebezium in sink Docker image*3.0.5.Final*Runtime classesDebezium source connector*3.4.0.Final*Producing the CDC eventsKafka Connect API (connector)*3.6.0* (compiled)Kafka Connect runtime*3.9.0* > The 
AddOpFieldForDebeziumRecord
uses 
SmtManager
from Debezium: >
Copy code
// line 51
> private SmtManager<R> smtManager;
> // line 66
> if (!smtManager.isValidEnvelope(record)) {
>     return record;  // ← Also silently returns if envelope validation fails
> }
> 
SmtManager
is from 
debezium-core:1.8.1.Final
(compile time) but at runtime Debezium 3.0.5 classes are on the classpath. The 
SmtManager
API has changed significantly between 1.8.x and 3.0.x — the 
isValidEnvelope()
method behavior may differ, potentially causing additional silent pass-throughs. Thanks to Claude
Now the issue is that the amount of the data on Kafka is gonna increase drastically, as for every single event, we have to include the schema, sth like:
Copy code
{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int64",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "wallet_deducted_amount_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "original_items_price_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "items_price_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "delivery_fee_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "basket_discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "items_discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "delivery_discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "default": 0,
            "field": "earned_cashback_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "vat_rate"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "vat_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "total_price_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "payable_amount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "paid_amount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "refunded_amount_cents"
          },
          {
            "type": "string",
            "optional": false,
            "field": "currency"
          },
          {
            "type": "string",
            "optional": false,
            "name": "io.debezium.time.ZonedTimestamp",
            "version": 1,
            "default": "1970-01-01T00:00:00Z",
            "field": "created_at"
          },
          {
            "type": "string",
            "optional": false,
            "name": "io.debezium.time.ZonedTimestamp",
            "version": 1,
            "default": "1970-01-01T00:00:00Z",
            "field": "updated_at"
          }
        ],
        "optional": true,
        "name": "dbz.orders.orders.order_charges.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int64",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "wallet_deducted_amount_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "original_items_price_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "items_price_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "delivery_fee_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "basket_discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "items_discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "delivery_discount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "default": 0,
            "field": "earned_cashback_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "vat_rate"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "vat_cents"
          },
          {
            "type": "int32",
            "optional": false,
            "default": 0,
            "field": "total_price_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "payable_amount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "paid_amount_cents"
          },
          {
            "type": "int32",
            "optional": true,
            "field": "refunded_amount_cents"
          },
          {
            "type": "string",
            "optional": false,
            "field": "currency"
          },
          {
            "type": "string",
            "optional": false,
            "name": "io.debezium.time.ZonedTimestamp",
            "version": 1,
            "default": "1970-01-01T00:00:00Z",
            "field": "created_at"
          },
          {
            "type": "string",
            "optional": false,
            "name": "io.debezium.time.ZonedTimestamp",
            "version": 1,
            "default": "1970-01-01T00:00:00Z",
            "field": "updated_at"
          }
        ],
        "optional": true,
        "name": "dbz.orders.orders.order_charges.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "connector"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_ms"
          },
          {
            "type": "string",
            "optional": true,
            "name": "io.debezium.data.Enum",
            "version": 1,
            "parameters": {
              "allowed": "true,first,first_in_data_collection,last_in_data_collection,last,false,incremental"
            },
            "default": "false",
            "field": "snapshot"
          },
          {
            "type": "string",
            "optional": false,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "sequence"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "ts_us"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "ts_ns"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "query"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "version": 1,
        "field": "source"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "total_order"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "data_collection_order"
          }
        ],
        "optional": true,
        "name": "event.block",
        "version": 1,
        "field": "transaction"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_us"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ns"
      }
    ],
    "optional": false,
    "name": "dbz.orders.orders.order_charges.Envelope",
    "version": 2
  },
  "payload": {
    "before": {
      "id": 115945775,
      "wallet_deducted_amount_cents": 0,
      "original_items_price_cents": 10000,
      "items_price_cents": 10000,
      "delivery_fee_cents": 0,
      "discount_cents": 0,
      "basket_discount_cents": 0,
      "items_discount_cents": 0,
      "delivery_discount_cents": 0,
      "earned_cashback_cents": 0,
      "vat_rate": 19,
      "vat_cents": 0,
      "total_price_cents": 10000,
      "payable_amount_cents": 10000,
      "paid_amount_cents": 0,
      "refunded_amount_cents": 0,
      "currency": "USD",
      "created_at": "2026-02-19T12:50:32Z",
      "updated_at": "2026-02-19T12:50:32Z"
    },
    "after": {
      "id": 115945775,
      "wallet_deducted_amount_cents": 0,
      "original_items_price_cents": 10000,
      "items_price_cents": 10000,
      "delivery_fee_cents": 0,
      "discount_cents": 0,
      "basket_discount_cents": 0,
      "items_discount_cents": 0,
      "delivery_discount_cents": 0,
      "earned_cashback_cents": 0,
      "vat_rate": 19,
      "vat_cents": 0,
      "total_price_cents": 10000,
      "payable_amount_cents": 10000,
      "paid_amount_cents": 10000,
      "refunded_amount_cents": 0,
      "currency": "USD",
      "created_at": "2026-02-19T12:50:32Z",
      "updated_at": "2026-02-19T12:50:46Z"
    },
    "source": {
      "version": "3.4.0.Final",
      "connector": "mysql",
      "name": "dbz.orders",
      "ts_ms": 1771505445512,
      "snapshot": "false",
      "db": "orders",
      "sequence": null,
      "ts_us": 1771505445512135,
      "ts_ns": 1771505445512135000,
      "table": "order_charges",
      "server_id": 1866973752,
      "gtid": "4cc4899c-b939-3272-8c32-41c597973732:5386034557",
      "file": "mysql-bin-changelog.070150",
      "pos": 2806445,
      "row": 0,
      "thread": 233048,
      "query": null
    },
    "transaction": null,
    "op": "u",
    "ts_ms": 1771505450426,
    "ts_us": 1771505450426459,
    "ts_ns": 1771505450426460000
  }
}
Open in Slack
PreviousNext
Powered by