Serializer error when inserting in table with JSON_SR

ksqlDB
1

I have this simple example json-schema:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://example.schema/test/v1",
  "title": "Test",
  "required": [
    "id",
    "values"
  ],
  "additionalProperties": false,
  "properties": {
    "id": {
      "type": "string",
      "minLength": 5
    },
    "values": {
      "type": "array",
      "minItems": 1,
      "uniqueItems": true,
      "items": {
        "type": "object",
        "additionalProperties": false,
        "required": [
          "key",
          "value"
        ],
        "properties": {
          "key": {
            "type": "integer",
            "minimum": 1
          },
          "value": {
            "type": "string",
            "pattern": "^.{3,}$"
          }
        }
      }
    }
  }
}

When I add this to schema-registry

curl -X POST http://localhost:8081/subjects/test-value/versions \
    -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    -d '{"schemaType": "JSON", "schema":"{\\"$schema\\":\\"https://json-schema.org/draft/2020-12/schema\\",\\"$id\\":\\"https://example.schema/test/v1\\",\\"title\\":\\"Test\\",\\"required\\":[\\"id\\",\\"values\\"],\\"additionalProperties\\":false,\\"properties\\":{\\"id\\":{\\"type\\":\\"string\\",\\"minLength\\":5},\\"values\\":{\\"type\\":\\"array\\",\\"minItems\\":1,\\"uniqueItems\\":true,\\"items\\":{\\"type\\":\\"object\\",\\"additionalProperties\\":false,\\"required\\":[\\"key\\",\\"value\\"],\\"properties\\":{\\"key\\":{\\"type\\":\\"integer\\",\\"minimum\\":1},\\"value\\":{\\"type\\":\\"string\\",\\"pattern\\":\\"^.{3,}$\\"}}}}}}"}'

I get a schema id back {"id":1}

I create a table using ksqldb-cli that references the schema.

CREATE TABLE `test` (`key` STRING PRIMARY KEY) WITH (kafka_topic='test', replicas=1, partitions=1, value_schema_id=1, value_format='JSON_SR');

This is the table

ksql> describe `test` extended;

Name                 : test
Type                 : TABLE
Timestamp field      : Not set - using <ROWTIME>
Key format           : KAFKA
Value format         : JSON_SR
Kafka topic          : test (partitions: 1, replication: 1)
Statement            : CREATE TABLE `test` (`key` STRING PRIMARY KEY, `values` ARRAY<STRUCT<`value` STRING, `key` BIGINT>>, `id` STRING) WITH (CLEANUP_POLICY='compact', KAFKA_TOPIC='test', KEY_FORMAT='KAFKA', PARTITIONS=1, REPLICAS=1, VALUE_FORMAT='JSON_SR', VALUE_SCHEMA_FULL_NAME='Test', VALUE_SCHEMA_ID=1);

 Field  | Type
-----------------------------------------------------------
 key    | VARCHAR(STRING)  (primary key)
 values | ARRAY<STRUCT<value VARCHAR(STRING), key BIGINT>>
 id     | VARCHAR(STRING)
-----------------------------------------------------------

When I try to insert a value to the table I get a serialization error:

ksql> INSERT INTO `test` (`key`, `id`, `values`) VALUES ('12345', '12345', ARRAY[STRUCT(`key`:=1, `value`:='abc')]);
Failed to insert values into 'test'. Could not serialize valueError serializing message to topic: test. Struct schemas do not match.
Hint: You probably forgot to add VALUE_SCHEMA_ID when creating the source.

How can I insert into the table?

2

I believe I’ve run into this myself, but under a different context.

You’ve created a schema using the JSON schema type, but the table uses JSON_SR, which is a different format.

Try it with the same format, if possible, and let me know if that works or if the error changes.

3

This is intentional. I want to use schema registry therefore I set it to JSON_SR and specify the value_schema_id I get from schema-registry.

Schema-registry only supports Json, Avro and Protobuf as schema types.

4

This topic was automatically closed after 30 days. New replies are no longer allowed.