Can't do update query via Kafka Connector.

Question

question

arsen.gasparyan.05_188257 asked · Jun 09 at 01:51 PM

Can't do update query via Kafka Connector.

I have the following Table

CREATE TABLE test.test ( id int, change_log list<text>, PRIMARY KEY (id) ) ;

Value : {"changeLog":["text"],"id":1234}

When I want to run update query by Kafka connector

topic.Test.test.test.query=UPDATE test.test SET change_log = change_log +  :changeLog WHERE id = :id ;

I get the following issue

ERROR Problem when getting queryFuture. This is likely a bug in the connector, please report. (com.datastax.kafkaconnector.DseSinkTask:153)
java.util.concurrent.ExecutionException: com.datastax.oss.driver.api.core.servererrors.ServerError: java.lang.ClassCastException: org.apache.cassandra.cql3.Constants$Value cannot be cast to org.apache.cassandra.cql3.Lists$Value

cassandra connector cql collections kafka

10 |1000 characters needed characters left characters exceeded

Attachments: Up to 8 attachments (including images) can be used with a maximum of 1.0 MiB each and 10.0 MiB total.

Answer 1 · 2020-06-10T08:26:37Z

Tomasz Lelek answered · Jun 10 at 08:26 AM

Please take a look at the logs that are produced by the Kafka connector.

When the prepared statement is produced you should be able to see the log starting with:

[2020-06-10 10:19:13,879] ERROR WorkerSinkTask{id=dse-sink-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:179)
java.lang.RuntimeException: Prepare failed for statement: UPDATE my_ks.test SET change_log = change_log + [:changeLog] WHERE id = :id ;

When you will scroll down at the end of the stack trace of this log you will see the underlying cause of your problems:

Caused by: com.datastax.oss.driver.api.core.servererrors.InvalidQueryException: Invalid list literal for change_log: bind variables are not supported inside collection literals

The error denotes that you cannot use bind variables (:changeLog) for collections.

It’s a limitation of the CQL itself.

I am enclosing the full exception log for a reference: Invalid_list_literal_error.txt

invalid-list-literal-error.txt (4.2 KiB)

Share

10 |1000 characters needed characters left characters exceeded

Attachments: Up to 8 attachments (including images) can be used with a maximum of 1.0 MiB each and 10.0 MiB total.

Answer 2 · 2020-06-10T08:00:08Z

alex.ott answered · Jun 10 at 08:00 AM

Unfortunately, CQL doesn't support binding inside the collection. If you do similar query inside the Java driver, you can get more detailed error, like:

com.datastax.driver.core.exceptions.InvalidQueryException: Invalid set literal for l: bind variables are not supported inside collection literals

To perform what you need to you need to modify your topic values to wrap that :changeLog value into a list instead of the single value, and then use query like this:

topic.Test.test.test.query=UPDATE test.test SET change_log = change_log + :changeLog WHERE id = :id ;

Transformation of single value into a list/array theoretically could be done with KSQL, or maybe custom transformations inside Kafka Connect framework.

P.S. I answered similar question on StackOverflow yesterday.

2 Share

10 |1000 characters needed characters left characters exceeded

Attachments: Up to 8 attachments (including images) can be used with a maximum of 1.0 MiB each and 10.0 MiB total.

arsen.gasparyan.05_188257 · Jun 10 at 08:58 AM

Anyway with list also I am reseving the same error

{"changeLog":["Hkf"],"id":178}

topic.Test.test.test.query=UPDATE test.test SET change_log = change_log + :changeLog WHERE id = :id ;

This is likely a bug in the connector, please report. (com.datastax.kafkaconnector.DseSinkTask:153)

java.util.concurrent.ExecutionException: com.datastax.oss.driver.api.core.servererrors.ServerError: java.lang.ClassCastException: org.apache.cassandra.cql3.Constants$Value cannot be cast to org.apache.cassandra.cql3.Lists$Value

0 · ·

alex.ott arsen.gasparyan.05_188257 · Jun 10 at 10:45 AM

You don't have list in your Kafka topic - you have single value that you need to transform to list using one of the methods I described (custom transformations, or KSQL).

0 · ·

Answer 3 · 2020-06-09T20:30:16Z

smadhavan answered · Jun 09 at 08:30 PM

@arsen.gasparyan.05_188257, it appears that the table schema expect change_log column value to be a list<text>`` whereas the actual JSON structure has it as text type: {..."changeLog":"text"...}. How about if you fix the JSON to have an actual list of strings something like {..."changeLog":["text"]...}?

Couple other references,

2 Share

10 |1000 characters needed characters left characters exceeded

Attachments: Up to 8 attachments (including images) can be used with a maximum of 1.0 MiB each and 10.0 MiB total.

arsen.gasparyan.05_188257 · Jun 09 at 09:58 PM

I also try as you mentioned and I have the following

ERROR Problem when getting queryFuture. This is likely a bug in the connector, please report. (com.datastax.kafkaconnector.DseSinkTask:153)

java.util.concurrent.ExecutionException: com.datastax.oss.driver.api.core.servererrors.ServerError: java.lang.ClassCastException: org.apache.cassandra.cql3.Constants$Value cannot be cast to org.apache.cassandra.cql3.Lists$Value

0 · ·

alexandre.dutra ♦ arsen.gasparyan.05_188257 · Jun 10 at 01:19 PM

This error seems strange, can you explain a bit more? It seems that the query you provided does not match the data you sent.

0 · ·

question