Flink - Cassandra sink - No support for the type of the given DataStream: Row<...>

I have a Flink (1.14, Java, Table API) app that consumes data from Kafka (Confluent Avro format), makes some transformation/aggregation and writes back to another Kafka topic (upsert-kafka connector).

Now I want to change sink to write results to Cassandra cluster (upsert). There is no Cassandra connector for the Table API. I'm trying to convert Table to DataStream and then use Apache Cassandra Connector (for DataStream API). But I'm getting an error.

Any ideas how to make it work?

My code:

    public static void main(String[] args) throws Exception {
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        ...
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        ...
        tableEnv.executeSql("CREATE TABLE `from_kafka_my_topic` (\n" +
                "    customer_id            STRING, \n" +
                "    service_id             STRING, \n" +
                "    upload_bytes           STRING, \n" +
                "    download_bytes         STRING, \n" +
                "    event_time_as_string   STRING, \n" +
                "    event_time AS TO_TIMESTAMP(event_time_as_string, 'yyyyMMddHHmmssX'), \n" +
                "    WATERMARK FOR event_time AS event_time - INTERVAL '10' MINUTE\n" +
                ") WITH (\n" +
                "    'connector'                            = 'kafka',\n" +
                "    'topic'                                = 'my_topic',\n" +
                "    'format'                               = 'avro-confluent',\n" +
                "    'scan.startup.mode'                    = '"+SOURCE__KAFKA_OFFSET_RESET+"-offset',\n" +
                "    'avro-confluent.schema-registry.url'   = '"+SOURCE__KAFKA_SCHEMA_REGISTRY_URL+"',\n" +
                "    'properties.group.id'                  = '"+SOURCE__KAFKA_GROUP_ID+"',\n" +
                "    'properties.bootstrap.servers'         =  '"+SOURCE__KAFKA_BOOTSTRAP_SERVERS+"'\n" +
                ")");
                
        Table resultTable = tableEnv.sqlQuery(
        " SELECT \n"+
                "     window_start \n"+
                "   , window_end \n"+
                "   , cast('DATA' as string) as record_type \n"+
                "   , COALESCE(substr(customer_id, 3), '0') customer_id \n"+
                "   , service_id \n"+
                "   , cast(null as bigint) duration \n"+
                "   , sum(cast(upload_bytes as bigint)+cast(download_bytes as bigint)) as volume \n"+
                " FROM TABLE ( \n"+
                "   TUMBLE(TABLE from_kafka_my_topic, DESCRIPTOR(event_time), INTERVAL '5' MINUTES) \n"+
                " ) \n"+
                " WHERE \n"+
                "   service_id is not null \n"+
                "   and coalesce(event_time_as_string, '') <> '' \n"+
                "   and cast(upload_bytes as bigint)+cast(download_bytes as bigint) > 0 \n"+
                " GROUP BY \n"+
                "     window_start \n"+
                "   , window_end \n"+
                "   , COALESCE(substr(customer_id, 3), '0') \n"+
                "   , service_id \n"+
        "");
        
        DataStream<Row> dataStream = tableEnv.toDataStream(resultTable);
        CassandraSink<Row> sink = CassandraSink
                .addSink(dataStream)
                .setClusterBuilder(new ClusterBuilder() {
                    @Override
                    protected Cluster buildCluster(Cluster.Builder builder) {
                        return Cluster.builder()
                                .addContactPoints("cassandra01.myhost.com,cassandra02.myhost.com".split(","))
                                .withPort(9042)
                                .withCredentials("LoginCass","123456")
                                .build();
                    }
                })
                .setQuery("INSERT INTO customers_usg.data_usage(window_start, window_end, record_type, customer_id, service_id, duration, volume) values (?, ?, ?, ?, ?, ?, ?);")
                .build();
        
        sink.name("Sink_to_Cassandra").disableChaining().uid("my_uid");
        
    }
}

Error:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: No support for the type of the given DataStream: ROW<window_start TIMESTAMP(3) NOT NULL, window_end TIMESTAMP(3) NOT NULL, record_type STRING NOT NULL, customer_id STRING NOT NULL, service_id STRING NOT NULL, duration BIGINT, volume BIGINT> NOT NULL(org.apache.flink.types.Row, org.apache.flink.table.runtime.typeutils.ExternalSerializer) at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114) at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246) at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054) at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1657) at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)