ClassCastException while using outputformat json

[rakulkar-tibco]

I am using cassandra 4 and apache pulsar 2.10.0. With cdc-apache-cassandra release 2.2.2 I have created following source

pulsar-admin source create \
--name cassandra-source-json \
--archive /home/softwares/cassandra-source-connectors-2.2.2/pulsar-cassandra-source-2.2.2.nar \
--tenant public \
--namespace default \
--destination-topic-name public/default/data-ks1.table1 \
--parallelism 1 \
--source-config '{
    "events.topic": "persistent://public/default/events-ks1.table1",
    "keyspace": "ks1",
    "table": "table1",
    "contactPoints": "localhost",
    "port": 9042,
    "loadBalancing.localDc": "datacenter1",
    "auth.provider": "PLAIN",
    "auth.username": "xyz",
    "auth.password": "xyz",
    "outputFormat" : "json"
}'

I am able to see events getting published on events-ks1.table1 , but not able to see them on data-ks1.table1.

I see following exception in connector logs:

ERROR org.apache.pulsar.client.impl.ProducerImpl - [public/default/data-ks1.table1] [null] Failed to create producer: class org.apache.pulsar.client.impl.schema.SchemaInfoImpl cannot be cast to class org.apache.pulsar.client.impl.schema.SchemaInfoImpl (org.apache.pulsar.client.impl.schema.SchemaInfoImpl is in unnamed module of loader org.apache.pulsar.common.nar.NarClassLoader @365a6a43; org.apache.pulsar.client.impl.schema.SchemaInfoImpl is in unnamed module of loader java.net.URLClassLoader @5451c3a8)
2023-01-25T23:34:41,603+0530 [public/default/cassandra-source-json-0] ERROR org.apache.pulsar.functions.sink.PulsarSink - Failed to create Producer while doing user publish
org.apache.pulsar.client.api.PulsarClientException: java.util.concurrent.ExecutionException: java.lang.ClassCastException: class org.apache.pulsar.client.impl.schema.SchemaInfoImpl cannot be cast to class org.apache.pulsar.client.impl.schema.SchemaInfoImpl (org.apache.pulsar.client.impl.schema.SchemaInfoImpl is in unnamed module of loader org.apache.pulsar.common.nar.NarClassLoader @365a6a43; org.apache.pulsar.client.impl.schema.SchemaInfoImpl is in unnamed module of loader java.net.URLClassLoader @5451c3a8)
    at org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:1074) ~[java-instance.jar:?]
    at org.apache.pulsar.client.impl.ProducerBuilderImpl.create(ProducerBuilderImpl.java:88) ~[org.apache.pulsar-pulsar-client-original-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.sink.PulsarSink$PulsarSinkProcessorBase.createProducer(PulsarSink.java:145) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.sink.PulsarSink$PulsarSinkProcessorBase.lambda$getProducer$0(PulsarSink.java:157) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1705) ~[?:?]
    at org.apache.pulsar.functions.sink.PulsarSink$PulsarSinkProcessorBase.getProducer(PulsarSink.java:153) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.sink.PulsarSink$PulsarSinkProcessorBase.getProducer(PulsarSink.java:149) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.sink.PulsarSink$PulsarSinkAtMostOnceProcessor.newMessage(PulsarSink.java:243) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.sink.PulsarSink.write(PulsarSink.java:365) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.instance.JavaInstanceRunnable.sendOutputMessage(JavaInstanceRunnable.java:380) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.instance.JavaInstanceRunnable.handleResult(JavaInstanceRunnable.java:363) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:304) ~[org.apache.pulsar-pulsar-functions-instance-2.10.2.jar:2.10.2]
    at java.lang.Thread.run(Thread.java:829) ~[?:?]
Caused by: java.util.concurrent.ExecutionException: java.lang.ClassCastException: class org.apache.pulsar.client.impl.schema.SchemaInfoImpl cannot be cast to class org.apache.pulsar.client.impl.schema.SchemaInfoImpl (org.apache.pulsar.client.impl.schema.SchemaInfoImpl is in unnamed module of loader org.apache.pulsar.common.nar.NarClassLoader @365a6a43; org.apache.pulsar.client.impl.schema.SchemaInfoImpl is in unnamed module of loader java.net.URLClassLoader @5451c3a8)
    at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395) ~[?:?]
    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999) ~[?:?]
    at org.apache.pulsar.client.impl.ProducerBuilderImpl.create(ProducerBuilderImpl.java:86) ~[org.apache.pulsar-pulsar-client-original-2.10.2.jar:2.10.2]
    ... 11 more

Could you please help?

[rakulkar-tibco]

I could get it working with Pulsar 2.8.3. But it is not working with higher pulsar versions.

[aymkhalil]

@rakulkar-tibco this is been fixed with #101 . I'll cut a new release 2.2.3v that will contain the fix today.

[aymkhalil]

Please use the cdc-apache-cassandra v2.2.3 release.