Fix ClassCastException by reverting NativeSchemaWrapper and CassandraSource to original implementation

- Reverted NativeSchemaWrapper.encode() to simple pass-through (return bytes)
- Reverted CassandraSource JsonValueRecord.getValue() and getKey() to original cast
- Root cause: Cannot handle GenericRecord in encode() due to method signature forcing byte[] cast
- The original simple implementation works correctly with Pulsar's internal handling

Author: Madhavan

commons/src/main/java/com/datastax/oss/cdc/NativeSchemaWrapper.java

@@ -16,26 +16,16 @@
 package com.datastax.oss.cdc;
 
 import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericDatumWriter;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.io.BinaryEncoder;
-import org.apache.avro.io.EncoderFactory;
 import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Optional;
 
 public class NativeSchemaWrapper implements org.apache.pulsar.client.api.Schema<byte[]> {
 
-    private static final Logger log = LoggerFactory.getLogger(NativeSchemaWrapper.class);
-
     private final SchemaInfo pulsarSchemaInfo;
     private final Schema nativeSchema;
 
@@ -53,41 +43,8 @@ public NativeSchemaWrapper(Schema nativeSchema, SchemaType pulsarSchemaType) {
     }
 
     @Override
-    public byte[] encode(byte[] data) {
-        if (data == null) {
-            throw new IllegalArgumentException("Cannot encode null data");
-        }
-
-        // The parameter is declared as byte[] to match the Schema<byte[]> interface,
-        // but Pulsar internally may pass GenericRecord objects. We need to handle both cases.
-        Object actualData = (Object) data;
-
-        // Handle byte[] input (backward compatibility)
-        if (actualData instanceof byte[]) {
-            return (byte[]) actualData;
-        }
-
-        // Handle GenericRecord input (Pulsar internal usage)
-        if (actualData instanceof GenericRecord) {
-            try {
-                GenericRecord record = (GenericRecord) actualData;
-                ByteArrayOutputStream out = new ByteArrayOutputStream();
-                BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
-                GenericDatumWriter<GenericRecord> writer = new GenericDatumWriter<>(nativeSchema);
-                writer.write(record, encoder);
-                encoder.flush();
-                return out.toByteArray();
-            } catch (IOException e) {
-                log.error("Failed to serialize GenericRecord to bytes", e);
-                throw new RuntimeException("Failed to serialize GenericRecord", e);
-            }
-        }
-
-        // Unexpected type
-        throw new IllegalArgumentException(
-            "Unsupported data type for encoding: " + actualData.getClass().getName() +
-            ". Expected byte[] or GenericRecord."
-        );
+    public byte[] encode(byte[] bytes) {
+        return bytes;
     }
 
     @Override

connector/src/main/java/com/datastax/oss/pulsar/source/CassandraSource.java

@@ -999,27 +999,7 @@ private class JsonValueRecord implements CassandraRecord {
         @Override
         public byte[] getValue() {
             try {
-                Object value = kvRecord.getValue().getValue();
-                // Handle both byte[] (from NativeConverter) and GenericRecord (from GenericConverter)
-                if (value instanceof byte[]) {
-                    return (byte[]) value;
-                } else if (value instanceof org.apache.pulsar.client.api.schema.GenericRecord) {
-                    // GenericRecord from Pulsar - need to serialize to bytes
-                    org.apache.pulsar.client.api.schema.GenericRecord genericRecord =
-                        (org.apache.pulsar.client.api.schema.GenericRecord) value;
-                    
-                    // Use the converter to serialize the GenericRecord to bytes
-                    // The converter's fromConnectData method handles the serialization
-                    @SuppressWarnings("unchecked")
-                    Converter<byte[], org.apache.pulsar.client.api.schema.GenericRecord, ?, byte[]> converter =
-                        (Converter<byte[], org.apache.pulsar.client.api.schema.GenericRecord, ?, byte[]>)
-                        kvRecord.converterAndQueryFinal.converter;
-                    return converter.fromConnectData(genericRecord);
-                } else {
-                    throw new IllegalStateException("Unexpected value type: " +
-                        (value != null ? value.getClass().getName() : "null") +
-                        ". Expected byte[] or GenericRecord");
-                }
+                return (byte[]) kvRecord.getValue().getValue();
             } catch (Exception err) {
                 throw new RuntimeException(err);
             }
@@ -1033,30 +1013,12 @@ public Schema getSchema() {
         @Override
         public Optional<String> getKey() {
             Object key = kvRecord.getValue().getKey();
-            // Handle both byte[] and GenericRecord for keys
-            if (key instanceof byte[]) {
-                // returns a json string in plain text. E.g.: key:[{"a":"38878"}]
-                return Optional.of(new String((byte[])key, StandardCharsets.UTF_8));
-            } else if (key instanceof org.apache.pulsar.client.api.schema.GenericRecord) {
-                // GenericRecord from Pulsar - need to serialize to bytes
-                org.apache.pulsar.client.api.schema.GenericRecord genericRecord =
-                    (org.apache.pulsar.client.api.schema.GenericRecord) key;
-                
-                try {
-                    // Use the key converter to serialize the GenericRecord to bytes
-                    @SuppressWarnings("unchecked")
-                    Converter<byte[], org.apache.pulsar.client.api.schema.GenericRecord, ?, byte[]> converter =
-                        (Converter<byte[], org.apache.pulsar.client.api.schema.GenericRecord, ?, byte[]>) keyConverter;
-                    byte[] keyBytes = converter.fromConnectData(genericRecord);
-                    return Optional.of(new String(keyBytes, StandardCharsets.UTF_8));
-                } catch (Exception e) {
-                    throw new RuntimeException("Failed to serialize key", e);
-                }
-            } else {
-                throw new IllegalStateException("Invalid key type: " +
-                    (key != null ? key.getClass().getName() : "null") +
-                    ". Expected byte[] or GenericRecord");
+            if (!(key instanceof byte[])) {
+                throw new IllegalStateException("Invalid key type " + key.getClass().getName());
             }
+
+            // returns a json string in plain text. E.g.: key:[{"a":"38878"}]
+            return Optional.of(new String((byte[])key, StandardCharsets.UTF_8));
         }
 
         @Override