DataException: Key must be a struct or map

Question

I am trying to configure the Confluent Cassandra Sink Connector with Kafka Connect using KSQLdb.

  CREATE SINK CONNECTOR cassandra WITH(
    "name" = 'CASSANDRA',
    "connector.class" = 'io.confluent.connect.cassandra.CassandraSinkConnector',
    "tasks.max" = '1',
    "topics" = 'users',
    "cassandra.contact.points" = 'cassandra',
    "cassandra.keyspace" = 'test',
    "confluent.topic.bootstrap.servers" = 'kafka:29092',
    "confluent.topic.replication.factor" = '1',
    "key.converter" = 'org.apache.kafka.connect.storage.StringConverter',
    "value.converter" = 'org.apache.kafka.connect.json.JsonConverter',
    "transforms" = 'createKey,extractId',
    "transforms.createKey.type" = 'org.apache.kafka.connect.transforms.ValueToKey',
    "transforms.createKey.fields" = 'ID',
    "transforms.extractId.type" = 'org.apache.kafka.connect.transforms.ExtractField$Key',
     "transforms.extractId.field" = 'ID');

The Users topic is a KSQLdb table.

Printing the topic; results are :

Key format: KAFKA_STRING

Value format: JSON or KAFKA_STRING

rowtime: 2020/05/19 10:51:35.036 Z, key: P343434, value: {"ID":"P343434"}

Exception :

Caused by: org.apache.kafka.connect.errors.DataException: Key must be a struct or map. This connector requires that records from Kafka contain the keys for the Cassandra table. Please use a transformation like org.apache.kafka.connect.transforms.ValueToKey to create a key with the proper fields.

The exception mentions that the key must be a struct or map !!! I did the transformation to give it a key, but the problem still there!

Is there any way to troubleshoot this or to know the required key/value format for this Cassandra connector?

Alex_Ott_18627 · Answer

The Confluent's Cassandra connector has following limitations (very big from my point of view):

Key of the topic must directly match to the primary key of the Cassandra table. So in your case you Cassandra table should have primary key consisting of one column with type text. If you have composite primary key, then you need to transform your topic into another, with struct or map matching the primary key in Cassandra
Value of the topic should match to the "regular" columns of table (everything except primary key)

This limitation lead to proliferation of the intermediate topics, etc.

The more flexible solution could be to use Kafka Sink Connector from DataStax:

It doesn't have such restrictions - you can define how you map your topic into table's fields
It's lightweight and high perfomant
Work with both DSE & Cassandra
Supports writing from single topic into multiple tables without need to create intermediate topic (as required in Confluent's version)
...

DataException: Key must be a struct or map

1 Answers