[GitHub] [pulsar] eolivelli commented on a change in pull request #10113: [Pulsar IO] Clarify usage of KVRecord on Sinks

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

eolivelli commented on a change in pull request #10113:
URL: https://github.com/apache/pulsar/pull/10113#discussion_r605890676



##########
File path: pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
##########
@@ -342,7 +345,14 @@ private void sendOutputMessage(Record srcRecord, Object output) {
             Thread.currentThread().setContextClassLoader(functionClassLoader);
         }
         try {
-            this.sink.write(new SinkRecord<>(srcRecord, output));
+            SinkRecord sinkRecord;

Review comment:
       @sijie 
   these are the key points:
   1) on the Source side you use KVRecord for records with a KeyValue payload, this lets you specify the keySchema and the valueSchema, for consistency it is good to have the same situation on the Sink side
   2) KeyValueSchema is not part of the public api (it is in the impl package), so in theory you cannot get the keySchema and the valueSchema when you are inside a Sink, if we implement KVRecord then you have those schemas
   3) When we have AutoConsumeSchema that returns a KeyValue object the record.getSchema() is the internal class AutoConsumeSchema, so you cannot still get the keySchema and the valueSchema  (this problem is still not present, because we have not merged the Sink<GenericObject patch yet
   
    




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

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