Kinesis Sink Cannot be Created via TableDescriptor

Caused by: java.lang.UnsupportedOperationException
    at java.base/java.util.Collections$UnmodifiableMap.remove(Collections.java:1460)
    at org.apache.flink.connector.kinesis.table.util.KinesisStreamsConnectorOptionsUtils$KinesisProducerOptionsMapper.removeMappedOptions(KinesisStreamsConnectorOptionsUtils.java:249)
    at org.apache.flink.connector.kinesis.table.util.KinesisStreamsConnectorOptionsUtils$KinesisProducerOptionsMapper.mapDeprecatedClientOptions(KinesisStreamsConnectorOptionsUtils.java:158)
    at org.apache.flink.connector.kinesis.table.util.KinesisStreamsConnectorOptionsUtils.<init>(KinesisStreamsConnectorOptionsUtils.java:90)
    at org.apache.flink.connector.kinesis.table.KinesisDynamicTableSinkFactory.createDynamicTableSink(KinesisDynamicTableSinkFactory.java:61)
    at org.apache.flink.table.factories.FactoryUtil.createDynamicTableSink(FactoryUtil.java:267)
    ... 20 more

public class Job {
  public static void main(String[] args) throws Exception {
    // create data stream environment
    StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    sEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING);
    StreamTableEnvironment tEnv = StreamTableEnvironment.create(sEnv);

    Schema a = Schema.newBuilder().column("a", DataTypes.STRING()).build();
    tEnv.createTemporaryTable(
        "exampleTable", TableDescriptor.forConnector("datagen").schema(a).build());
    TableDescriptor descriptor =
        TableDescriptor.forConnector("kinesis")
            .schema(a)
            .format("json")
            .option("stream", "abc")
            .option("aws.region", "eu-central-1")
            .build();
    tEnv.createTemporaryTable("sinkTable", descriptor);

    tEnv.from("exampleTable").executeInsert("sinkTable"); // error occurs here
  }
} 

tEnv.executeSql("CREATE TABLE sinkTable " + descriptor.toString());