search

datastax / pulsar-sink

An Apache Pulsar® sink for transferring events/messages from Pulsar topics to Apache Cassandra®, DataStax Astra or DataStax Enterprise (DSE) tables.
Apache License 2.0
14 stars 10 forks source link

fix: Error decoding/mapping Pulsar record with blob fields from CDC #61

Closed dlg99 closed 2 months ago

dlg99 commented 2 months ago

Fixes AS-4018

com.datastax.oss.sink.pulsar.CassandraSinkTask - Details of the error
java.lang.UnsupportedOperationException: type class java.nio.HeapByteBuffer
    at com.datastax.oss.sink.pulsar.PulsarSchema.of(PulsarSchema.java:85) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.sink.pulsar.PulsarSchema.update(PulsarSchema.java:144) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.sink.pulsar.PulsarSchema.<init>(PulsarSchema.java:101) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.sink.pulsar.PulsarSchema.createFromStruct(PulsarSchema.java:90) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.sink.pulsar.LocalSchemaRegistry.ensureAndUpdateSchema(LocalSchemaRegistry.java:58) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.sink.pulsar.PulsarStruct.ofRecord(PulsarStruct.java:68) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.sink.pulsar.PulsarSinkRecordImpl.unwrap(PulsarSinkRecordImpl.java:114) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.sink.pulsar.PulsarSinkRecordImpl.value(PulsarSinkRecordImpl.java:132) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.common.sink.AbstractSinkTask.mapAndQueueRecord(AbstractSinkTask.java:199) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at com.datastax.oss.common.sink.AbstractSinkTask.lambda$put$1(AbstractSinkTask.java:108) ~[cassandra-sink-pulsar-1.6.11.jar:?]
    at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) ~[?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[?:?]
    at java.lang.Thread.run(Thread.java:1570) ~[?:?]