Flink Cassandra No support for the type of the given DataStream: GenericType<scala.Tuple2>

I'm trying to write to Cassandra a list of tuple 2. Scala 2.11.12 Flink 1.12.1

import com.datastax.driver.core.Cluster
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.cassandra.{CassandraSink, ClusterBuilder, MapperOptions}

import java.util.Properties

object Kafka {
  def main(args: Array[String]) {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "broker:9092")
    properties.setProperty("group.id", "FlinkTest")
    implicit val typeInfo = TypeInformation.of(classOf[(String)])
    implicit val typeInfo_2 = TypeInformation.of(classOf[(String, Int)])
    val list = List(new Tuple2("a", 1), new Tuple2("b", 2), new Tuple2("c", 3))
    val sink: CassandraSink[(String, Int)] = CassandraSink.addSink(env.fromCollection(list))
      .setQuery("INSERT INTO flink_test.example (id, counter) values (?, ?);")
      .setClusterBuilder(new ClusterBuilder {
        override def buildCluster(builder: Cluster.Builder): Cluster = {
          builder.addContactPoint("cluster1").addContactPoint("cluster2").addContactPoint("cluster3").withCredentials("cassandra", "ABCXYZ123").build()
        }
      })
      .build()

    sink.name("sample").disableChaining.setParallelism(1).uid("hello")
    env.execute("Flink Kafka Example refresh")
  }
}

I got an error about type input

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: No support for the type of the given DataStream: GenericType<scala.Tuple2>
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:360)
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:213)
        at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:816)
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:248)
        at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1058)
        at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1136)
        at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1136)
Caused by: java.lang.IllegalArgumentException: No support for the type of the given DataStream: GenericType<scala.Tuple2>
        at org.apache.flink.streaming.connectors.cassandra.CassandraSink.addSink(CassandraSink.java:228)
        at org.apache.flink.streaming.connectors.cassandra.CassandraSink.addSink(CassandraSink.java:201)
        at Kafka$.main(Kafka.scala:18)
        at Kafka.main(Kafka.scala)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:566)
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:343)
        ... 8 more

I'm tried convert Datastream to javaStream by "env.fromCollection(list).javaStream" but this error still happen.

Why addSink function accept Datastream input type but this error happen?