Impossible to sync 2(two) kafka topics to same cassandra table

[mgergalov]

It's impossible to sync two kafka topics(trades-a,trades-b) in one cassandra table (trades).
If i try to sync it's separately (for example trades-a to trades) it work perfect.

that is props for separetely attemp:

connector.class=com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector
connect.cassandra.key.space=markets
topics=TRADES-a
tasks.max=1
connect.cassandra.kcql=INSERT INTO trade SELECT * FROM TRADES-a;
connect.cassandra.fetch.size=5000
connect.cassandra.max.retries=20
connect.progress.enabled=false
connect.cassandra.error.policy=throw
connect.cassandra.contact.points=docker.for.mac.localhost
connect.cassandra.port=9042
connect.cassandra.threadpool.size=10
name=CassandraSinkConnector
connect.cassandra.retry.interval=30000
connect.progress.enabled=true
connect.cassandra.consistency.level=LOCAL_ONE
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
key.converter.schemas.enable=false

As I Say it’s work fine.
And if i delete this connector and create TRADES-b to trade connector it’s work fine too!

but If I try to create this

connector.class=com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector
connect.cassandra.key.space=markets
topics=TRADES-a,TRADES-b
tasks.max=1
connect.cassandra.kcql=INSERT INTO trade SELECT * FROM TRADES-a;INSERT INTO trade SELECT * FROM TRADES-b;
connect.cassandra.fetch.size=5000
connect.cassandra.max.retries=20
connect.progress.enabled=false
connect.cassandra.error.policy=throw
connect.cassandra.contact.points=docker.for.mac.localhost
connect.cassandra.port=9042
connect.cassandra.threadpool.size=10
name=CassandraSinkConnector
connect.cassandra.retry.interval=30000
connect.progress.enabled=true
connect.cassandra.consistency.level=LOCAL_ONE
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
key.converter.schemas.enable=false

It’s doesn’t work..

TRACE:

org.apache.kafka.connect.errors.ConnectException: **No tables found in Cassandra for topics trades,trades**
	at com.datamountaineer.streamreactor.connect.cassandra.utils.CassandraUtils$.checkCassandraTables(CassandraUtils.scala:61)
	at com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraJsonWriter.<init>(CassandraJsonWriter.scala:54)
	at com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraWriter$.apply(CassandraWriter.scala:43)
	at com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkTask.start(CassandraSinkTask.scala:58)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:232)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:145)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:748)

Strange log..

I repeat, the error is obtained only if I make the insertion into the same table from different topics.
If I do an insertion into different tables from different topics, this works well

[andrewstevenson]

@mgergalov Give us 10 minutes, I see the issue.

[andrewstevenson]

@mgergalov Please try the latest from master and reopen if required

cd kafka-connect-cassandra
gradle fatJar