Handling KeyValue AVRO Messages in Pulsar with Incompatible Schema Exception

[sahatsawats]

Environment

• DataStax Enterprise 6.8.42 (Cassandra)
• Luna Streaming (Apache Pulsar version 2.7.2)

Problem Statement

I want to create a project that will received the CDC messages from Cassandra Cluster to Pulsar then perform some aggregation with time-window function.

[CDC Agent] --> [Events-topic] --> [Source Connector] --> [Data-topic] --> **[Function]** --> [Output-topic]

The Java function will received the KeyValue type with key and value are AVRO following the latest schema from the Data-topic. **(The latest schema generated by --st auto_consume option)

Latest Data-topic Schema:
{
  "version": 1,
  "schemaInfo": {
    "name": "data-exampleio.demo",
    "schema": {
      "key": {
        "name": "demo",
        "schema": {
          "type": "record",
          "name": "demo",
          "namespace": "exampleio",
          "doc": "Table exampleio.demo",
          "fields": [
            {
              "name": "id",
              "type": "int"
            }
          ]
        },
        "type": "AVRO",
        "timestamp": 0,
        "properties": {}
      },
      "value": {
        "name": "demo",
        "schema": {
          "type": "record",
          "name": "demo",
          "namespace": "exampleio",
          "doc": "Table exampleio.demo",
          "fields": [
            {
              "name": "amount",
              "type": [
                "null",
                {
                  "type": "record",
                  "name": "cql_decimal",
                  "namespace": "",
                  "fields": [
                    {
                      "name": "bigint",
                      "type": "bytes"
                    },
                    {
                      "name": "scale",
                      "type": "int"
                    }
                  ],
                  "logicalType": "cql_decimal"
                }
              ]
            },
            {
              "name": "context",
              "type": [
                "null",
                "string"
              ]
            }
          ]
        },
        "type": "AVRO",
        "timestamp": 0,
        "properties": {}
      }
    },
    "type": "KEY_VALUE",
    "timestamp": 1709620937104,
    "properties": {
      "key.schema.name": "demo",
      "key.schema.properties": "{}",
      "key.schema.type": "AVRO",
      "kv.encoding.type": "SEPARATED",
      "value.schema.name": "demo",
      "value.schema.properties": "{}",
      "value.schema.type": "AVRO"
    }
  }
}

However, my function is not working, after review the function log, the problem is the function-consumer is received the messages in bytes format and incompatible with the existing schema.

2024-03-05T12:53:18,038+0000 [public/default/demo_sumbyId-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [public/default/demo_sumbyId:0] Uncaught exception in Java Instance java.util.concurrent.CompletionException: org.apache.pulsar.client.api.PulsarClientException$IncompatibleSchemaException: {"errorMsg":"Key schemas or Value schemas are different schema type, from key schema type is AVRO and to key schema is BYTES, from value schema is AVRO and to value schema is BYTES","reqId":1870435977707154767, "remote":"Censored", "local":"Censored"}

In Client-CLI, I cannot receive the structure messages without the --st auto_consume even after the schema is registered.

with --st auto_consume:

key:[MA==], properties:[writetime=1709646681322536], content:{key={id=24}, value={amount={scale=2, bigint=java.nio.HeapByteBuffer[pos=0 lim=3 cap=3]}, context=exampleagain}}

without --st auto_consume:

key:[Mg==], properties:[writetime=1709647768862639], content:??exampleagain

Could you help identify the problem or what I missing please?