Github user ctamisier commented on the issue: https://github.com/apache/flink/pull/5964 > The keyspace field in the sink isn't used anywhere It is used in CassandraPojoSinkBuilder.createSink(), the keyspace is an argument of `new CassandraPojoSink<>(...);` ``` @Override public CassandraSink<IN> createSink() throws Exception { return new CassandraSink<>(input.addSink(new CassandraPojoSink<>(typeInfo.getTypeClass(), builder, mapperOptions, keyspace)).name("Cassandra Sink")); } ``` > and the configuration operations don't have any effect either the keyspace is set in the configuration in the `"keyspace"` key. ``` if (keyspace != null && !keyspace.isEmpty()) { configuration.setString("keyspace", keyspace); } ``` which is then used to connect to cassandra with this keyspace: `cluster.connect(configuration.getString("keyspace", null));` Am I missing something ? PS: Instead of using the configuration object maybe we can just use `session` that is just created to execute a `session.executeAsync("USE " + keyspace);` like in Cluster.connectAsync(final String keyspace) ?
---