[Bug] Kafka debezium-json cannot infer primary keys from standard Debezium message key

[TKilome]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

1.4.2, master

Compute Engine

Flink 1.20, kafka_sync_database

Minimal reproduce step

Use Debezium MySQL Connector to write standard Debezium JSON records to Kafka.

The Kafka record key contains the primary key fields, for example:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int64",
        "optional": false,
        "field": "id"
      }
    ],
    "optional": false,
    "name": "mysql_cluster.test.users.Key"
  },
  "payload": {
    "id": 1
  }
}

The Kafka record value is a standard Debezium JSON envelope and does not contain pkNames:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int64",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "name"
          }
        ],
        "optional": true,
        "field": "after"
      }
    ],
    "optional": false
  },
  "payload": {
    "before": null,
    "after": {
      "id": 1,
      "name": "Alice"
    },
    "source": {
      "db": "test",
      "table": "users"
    },
    "op": "c"
  }
}

Then run kafka_sync_database with value.format=debezium-json and a bucketed table configuration, for example:

kafka_sync_database \
   --warehouse s3://flink/paimon/warehouse \
   --database ods \
   --table_prefix ods_ \
   --kafka_conf properties.bootstrap.servers=localhost:9092 \
   --kafka_conf topic=cdc_mysql_cluster_test \
   --kafka_conf properties.group.id=paimon-kafka-cdc-test \
   --kafka_conf value.format=debezium-json \
   --kafka_conf scan.startup.mode=earliest-offset \
   --catalog_conf metastore=filesystem \
   --table_conf bucket=4 \
   --table_conf changelog-producer=input

What doesn't meet your expectations?

Paimon fails to infer primary keys from the standard Debezium JSON Kafka key.

KafkaDebeziumJsonDeserializationSchema already parses the Kafka key and stores it in CdcSourceRecord:

out.collect(new CdcSourceRecord(message.topic(), keyNode, valueNode, kafkaMetadata));

However, DebeziumJsonRecordParser inherits AbstractJsonRecordParser.extractPrimaryKeys(), which only reads primary keys from the JSON value field pkNames:

ArrayNode pkNames = getNodeAs(root, primaryField(), ArrayNode.class);

For Debezium JSON, primaryField() returns FIELD_PRIMARY, and FIELD_PRIMARY is pkNames.

Standard Debezium JSON does not put primary key field names in value pkNames. The primary key columns are represented by the Kafka message key schema/payload instead.

As a result, Paimon treats a primary-key source table as a table without primary keys. With bucketed table options, table creation may fail with:

You should define a 'bucket-key' for bucketed append mode.

And then downstream writer may fail to load the table:

Catalog$TableNotExistException: Table ods.ods_users does not exist

Expected behavior:

For standard Debezium JSON, Paimon should infer primary keys from the Kafka message key when value pkNames is absent.

Suggested behavior:

1. Keep current pkNames parsing for backward compatibility.
2. If pkNames is absent or empty, fallback to currentRecord.getKey().
3. For schema-enabled Debezium JSON key, extract primary key names from:

key.schema.fields[].field

4. For schema-disabled Debezium JSON key, extract primary key names from key object field names.

This would align Debezium JSON behavior with Debezium Avro behavior, where DebeziumAvroRecordParser.extractPrimaryKeys() already uses the Kafka key schema fields:

Schema keySchema = sanitizedSchema(keyRecord.getSchema());
return keySchema.getFields().stream().map(Schema.Field::name).collect(Collectors.toList());

Anything else?

This issue is about standard Debezium JSON records produced by Debezium MySQL Connector / Kafka Connect.

Debezium's message.key.columns option can customize which columns are used in the Kafka message key, but it does not add pkNames to the message value. Therefore Paimon should not
rely only on value pkNames for standard Debezium JSON.

Are you willing to submit a PR?

• I'm willing to submit a PR!