[GitHub] [pulsar] eolivelli commented on a diff in pull request #15025: [pulsar-io] KCA: handle whatever KeyValue getNativeObject() returns: org.apache.pulsar.io.core.KeyValue or org.apache.pulsar.common.schema.KeyValue

[GitBox <gi...@apache.org>]

eolivelli commented on code in PR #15025:
URL: https://github.com/apache/pulsar/pull/15025#discussion_r842060050


##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java:
##########
@@ -255,11 +254,25 @@ protected SinkRecord toSinkRecord(Record<GenericObject> sourceRecord) {
                 && sourceRecord.getSchema().getSchemaInfo() != null
                 && sourceRecord.getSchema().getSchemaInfo().getType() == SchemaType.KEY_VALUE) {
             KeyValueSchema kvSchema = (KeyValueSchema) sourceRecord.getSchema();
-            KeyValue kv = (KeyValue) sourceRecord.getValue().getNativeObject();
             keySchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getKeySchema());
             valueSchema = PulsarSchemaToKafkaSchema.getKafkaConnectSchema(kvSchema.getValueSchema());
-            key = kv.getKey();
-            value = kv.getValue();
+
+            Object nativeObject = sourceRecord.getValue().getNativeObject();
+
+            if (nativeObject instanceof org.apache.pulsar.common.schema.KeyValue) {
+                org.apache.pulsar.common.schema.KeyValue kv = (org.apache.pulsar.common.schema.KeyValue) nativeObject;
+                key = kv.getKey();
+                value = kv.getValue();

Review Comment:
   This is non possible.
   The wrapper for KeyValue is created in AutoConsumeSchema.
   
   This class is like a utility class in Pulsar IO. Probably there is some mock test that creates the wrong object.
   
   I believe we can safely drop this



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org