Context : We have a flink application that consumes data from kafka and sinks the data onto Scylla DB
When the application starts there is a Schema parsing error
Why are we getting this error ?
Flink Version : 1.15.4
Scylla DB version : scylla-enterprise-2022.2.11-0
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-cassandra_2.12</artifactId>
<version>1.15.4</version>
</dependency>
WARNING: All illegal access operations will be denied in a future release
2023-08-11 09:16:04.567 [Source: Custom Source -> Map -> Sink: Cassandra Sink (7/9)#0] ERROR com.datastax.driver.core.SchemaParser - Error parsing schema for table system_distributed.service_levels: Cluster.getMetadata().getKeyspace("system_distributed").getTable("service_levels") will be missing or incomplete
com.datastax.driver.core.exceptions.UnresolvedUserTypeException: Cannot resolve user type system_distributed.duration
at com.datastax.driver.core.DataTypeCqlNameParser.parse(DataTypeCqlNameParser.java:147)
at com.datastax.driver.core.TableMetadata.build(TableMetadata.java:188)
at com.datastax.driver.core.SchemaParser.buildTables(SchemaParser.java:176)
at com.datastax.driver.core.SchemaParser.buildKeyspaces(SchemaParser.java:128)
at com.datastax.driver.core.SchemaParser.refresh(SchemaParser.java:64)
at com.datastax.driver.core.ControlConnection.refreshSchema(ControlConnection.java:341)
at com.datastax.driver.core.ControlConnection.tryConnect(ControlConnection.java:271)
at com.datastax.driver.core.ControlConnection.reconnectInternal(ControlConnection.java:199)
at com.datastax.driver.core.ControlConnection.connect(ControlConnection.java:77)
at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1414)
Any input will be helpful
The Cannot resolve user type system_distributed.duration message indicates that the underlying driver is unaware of the duration type.
Upgrade the driver at least to v3.2.0, which is the initial version supporting durations. Notice that the latest Java driver is already at v4.17, so prefer switching to v4.x if Flink already supports the newer driver API.