[GitHub] [pulsar] dlg99 commented on a diff in pull request #19861: [fix][io] KCA sink: handle null values with KeyValue schema

["dlg99 (via GitHub)" <gi...@apache.org>]

dlg99 commented on code in PR #19861:
URL: https://github.com/apache/pulsar/pull/19861#discussion_r1142480456


##########
pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSinkTest.java:
##########
@@ -690,6 +694,20 @@ public void schemaKeyValueSchemaTest() throws Exception {
         Assert.assertEquals(key, 11);
     }
 
+    @Test
+    public void schemaKeyValueSchemaNullValueTest() throws Exception {
+        RecordSchemaBuilder builder = SchemaBuilder
+                .record("test");
+        builder.field("test").type(SchemaType.STRING);
+        GenericSchema<GenericRecord> schema = GenericAvroSchema.of(builder.build(SchemaType.AVRO));
+        KeyValue<Integer, String> kv = new KeyValue<>(11, null);
+        SinkRecord sinkRecord = recordSchemaTest(kv, Schema.KeyValue(Schema.INT32, schema), 11,
+                "INT32", null, "STRUCT");
+        Assert.assertNull(sinkRecord.value());
+        int key = (int) sinkRecord.key();
+        Assert.assertEquals(key, 11);
+    }
+

Review Comment:
   ```suggestion
   
       @Test
       public void schemaKeyValueSchemaNullValueNoUnwrapTest() throws Exception {
           props.put("unwrapKeyValueIfAvailable", "false");
           JSONSchema<PulsarSchemaToKafkaSchemaTest.StructWithAnnotations> jsonSchema = JSONSchema
                   .of(SchemaDefinition.<PulsarSchemaToKafkaSchemaTest.StructWithAnnotations>builder()
                           .withPojo(PulsarSchemaToKafkaSchemaTest.StructWithAnnotations.class)
                           .withAlwaysAllowNull(true)
                           .build());
   
           KeyValue<Integer, String> kv = new KeyValue<>(11, null);
   
           Map<String, Object> expected = new HashMap();
           expected.put("11", null);
           SinkRecord sinkRecord = recordSchemaTest(kv, Schema.KeyValue(Schema.INT32, jsonSchema), "key",
                   "STRING", expected, "MAP");
           Assert.assertNull(((Map)sinkRecord.value()).get(11));
           String key =(String)sinkRecord.key();
           Assert.assertEquals(key, "key");
       }
   
       @Test
       public void schemaKeyValueSchemaNullValueNoUnwrapTestAvro() throws Exception {
           props.put("unwrapKeyValueIfAvailable", "false");
           RecordSchemaBuilder builder = SchemaBuilder
                   .record("test");
           builder.property("op", "test");
           builder.field("test").type(SchemaType.STRING);
           GenericSchema<GenericRecord> schema = GenericAvroSchema.of(builder.build(SchemaType.AVRO));
           KeyValue<Integer, String> kv = new KeyValue<>(11, null);
   
           Map<String, Object> expected = new HashMap();
           expected.put("11", null);
   
           SinkRecord sinkRecord = recordSchemaTest(kv, Schema.KeyValue(Schema.INT32, schema), "key",
                   "STRING", expected, "MAP");
           Assert.assertNull(((Map)sinkRecord.value()).get(11));
           String key =(String)sinkRecord.key();
           Assert.assertEquals(key, "key");
       }
   
   ```



-- 
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