[
https://issues.apache.org/jira/browse/CASSSIDECAR-507?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134166#comment-18134166
]
Laxmikant Upadhyay commented on CASSSIDECAR-507:
------------------------------------------------
Hey [~Jyothsnakonisa] Regarding Avro schema generation issue :
I encountered two issues with the Avro schema generated for nested Cassandra
UDTs.
# *Invalid Avro namespace:* The generated namespace contains the raw CQL
collection type, such as {{{}list<frozen<execution_result>>{}}}, which includes
characters <> that are not valid in Avro identifiers.
# *Incorrect nullability:* Fields inside nested UDTs are generated as
non-nullable Avro types, even though Cassandra allows these UDT fields to
contain null values.
{code:java}
Create schema
CREATE TYPE IF NOT EXISTS cdc_demo.transient_event (
id uuid,
event_class text
);
CREATE TYPE IF NOT EXISTS cdc_demo.execution_result (
domain_event frozen<transient_event>,
error_reason text,
status text
);
CREATE TABLE IF NOT EXISTS cdc_demo.command (
id uuid PRIMARY KEY,
command_class text,
execution_results list<frozen<execution_result>>
) WITH cdc = true;
Insert a record with a null UDT field
INSERT INTO cdc_demo.command (
id,
command_class,
execution_results
) VALUES (
11111111-1111-1111-1111-111111111111,
'CreateOrderCommand',
[
{
domain_event: {
id: 22222222-2222-2222-2222-222222222222,
event_class: 'OrderCreated'
},
error_reason: null,
status: 'SUCCESS'
}
]
);
The Cassandra INSERT succeeds. Observe the message in KafkaUi you will see
registery can parse it. and the reason is the generated schema as below
{
"type": "record",
"name": "record_udt",
"namespace":
"cdc_demo.command.execution_results.list<frozen<execution_result>>.record_udt",
"fields": [
{
"name": "error_reason",
"type": {
"type": "string",
"cqlType": "text"
}
}
]
}
so the schema has two issues :
1. invaid <> chracters
2. the error_reason field is non nullable.
Ideally the schema may look like below ?
{
"type": "record",
"name": "record_udt",
"namespace": "cdc_demo.command.execution_results.execution_result",
"fields": [
{
"name": "error_reason",
"type": [
"null",
{
"type": "string",
"cqlType": "text"
}
],
"default": null
}
]
}
{code}
> CDC fails to deserialize nested UDT columns
> -------------------------------------------
>
> Key: CASSSIDECAR-507
> URL: https://issues.apache.org/jira/browse/CASSSIDECAR-507
> Project: Sidecar for Apache Cassandra
> Issue Type: Bug
> Components: Configuration
> Reporter: Laxmikant Upadhyay
> Assignee: Jyothsna Konisa
> Priority: Major
> Fix For: 0.5
>
> Time Spent: 40m
> Remaining Estimate: 0h
>
> While testing Cassandra CDC using the {{*docker/cdc-demo*}} \{*}setup from
> the Apache Cassandra Sidecar trunk branch{*}, with the locally built
> {{*cassandra-sidecar:dev*}} \{*}Docker image{*}, I encountered a
> {{ClassCastException}} when processing CDC mutations containing nested UDT
> columns.
> h3. Steps to Reproduce using docker compose cdc-demo
>
> {code:java}
> 1. Create a simple UDT
>
> {{CREATE TYPE IF NOT EXISTS cdc_demo.task_result (
> status text,
> message text
> );}}
> 2. Create a CDC-enabled table containing a list of frozen UDTs
>
> {{CREATE TABLE IF NOT EXISTS cdc_demo.task (
> id uuid PRIMARY KEY,
> name text,
> results list<frozen<task_result>>
> ) WITH cdc = true;}}
> 3. Insert a record containing the UDT collection
>
> {{INSERT INTO cdc_demo.task (
> id,
> name,
> results
> ) VALUES (
> 11111111-1111-1111-1111-111111111111,
> 'test-task',
> [
> { status: 'SUCCESS', message: 'completed' }
> ]
> );}}
> {code}
>
> *4. Observe Sidecar CDC logs*
> The CDC consumer detects the mutation and resolves the correct Kafka topic,
> but fails while deserializing the {{results}} column.
>
>
> {code:java}
> {{ERROR CdcLogMode.java:166 - Failed to deserialize column value.
> columnName=results
> java.lang.ClassCastException: null
> INFO CdcLogMode.java:104 - Processing CDC event.
> kind=INSERT keyspace=cdc_demo table=task
> partitionKeys=[id : 11111111-1111-1111-1111-111111111111]
> clusteringKeys=null
> valueColumns=[name : test-task], [results : unknown]
> topic=cdc-mutations-cdc_demo-task
> WARN CdcLogMode.java:113 - Skip publishing the event because it cannot be
> serialized.}}
> {code}
>
> *In a separate reproduction using a below schema, the exception included
> additional classloader details:*
>
> {code:java}
> – 1. Create transient_event UDT
> CREATE TYPE IF NOT EXISTS cdc_demo.transient_event (
> id uuid,
> event_class text
> );
> – 2. Create execution_result UDT
> CREATE TYPE IF NOT EXISTS cdc_demo.execution_result (
> domain_event frozen<transient_event>,
> status text
> );
> – 3. Create command table with CDC enabled
> CREATE TABLE IF NOT EXISTS cdc_demo.command (
> id uuid PRIMARY KEY,
> command_class text,
> execution_results list<frozen<execution_result>>
> ) WITH cdc = true;
>
> {{INSERT INTO cdc_demo.task (
> id,
> name,
> results
> ) VALUES (
> 11111111-1111-1111-1111-111111111111,
> 'test-task',
> [
> {
> status: 'SUCCESS',
> message: 'completed'
> }
> ]
> );}}
> {{}}
> {code}
>
> {{you will observe:}}
>
> {code:java}
> {{java.lang.ClassCastException:
> class org.apache.cassandra.spark.data.complex.CqlUdt
> cannot be cast to
> class org.apache.cassandra.spark.data.CqlType
> (org.apache.cassandra.spark.data.complex.CqlUdt
> is in unnamed module of loader
> org.apache.cassandra.bridge.PostDelegationClassLoader @7b1655c3;
> org.apache.cassandra.spark.data.CqlType
> is in unnamed module of loader
> org.apache.cassandra.bridge.PostDelegationClassLoader @38afe24f)}}
> {code}
>
> h3. Expected Behavior
> CDC should successfully deserialize UDT column values and publish the
> mutation to the configured Kafka topic.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
