cbornet commented on code in PR #16740:
URL: https://github.com/apache/pulsar/pull/16740#discussion_r954978419
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:
##########
@@ -383,11 +411,64 @@ void handleResult(Record srcRecord, JavaExecutionResult
result) throws Exception
private void sendOutputMessage(Record srcRecord, Object output) throws
Exception {
if (componentType ==
org.apache.pulsar.functions.proto.Function.FunctionDetails.ComponentType.SINK) {
- Thread.currentThread().setContextClassLoader(functionClassLoader);
+ Thread.currentThread().setContextClassLoader(componentClassLoader);
}
AbstractSinkRecord<?> sinkRecord;
if (output instanceof Record) {
- sinkRecord = new OutputRecordSinkRecord<>(srcRecord, (Record)
output);
+ Record record = (Record) output;
+ if (sinkSchemaInfoProvider != null) {
+ // Function and Sink coupled together so we need to encode
with the Function Schema
+ // and decode with the Sink schema
+ Schema encodingSchema = record.getSchema();
+ boolean isKeyValueSeparated = false;
+ if (encodingSchema instanceof KeyValueSchema) {
+ KeyValueSchema<?, ?> kvSchema = (KeyValueSchema<?, ?>)
encodingSchema;
+ // If the encoding is SEPARATED, it's easier to
encode/decode with INLINE
+ // and rebuild the SEPARATED KeyValueSchema after decoding
+ if (kvSchema.getKeyValueEncodingType() ==
KeyValueEncodingType.SEPARATED) {
+ encodingSchema =
KeyValueSchemaImpl.of(kvSchema.getKeySchema(), kvSchema.getValueSchema());
+ isKeyValueSeparated = true;
+ }
+ }
+ byte[] encoded = encodingSchema.encode(record.getValue());
+
+ if (sinkSchema.get() == null) {
+ Schema<?> schema = getSinkSchema(record, sinkTypeArg);
+ schema.setSchemaInfoProvider(sinkSchemaInfoProvider);
+ sinkSchema.compareAndSet(null, schema);
+ }
+ Schema<?> schema = sinkSchema.get();
+ SchemaVersion schemaVersion =
sinkSchemaInfoProvider.getSchemaVersion(encodingSchema);
+ final byte[] schemaVersionBytes = schemaVersion.bytes();
+ Object decoded = schema.decode(encoded, schemaVersionBytes);
+
+ if (schema instanceof AutoConsumeSchema) {
+ schema = ((AutoConsumeSchema)
schema).getInternalSchema(schemaVersionBytes);
+ }
+
+ final Schema<?> finalSchema;
+ if (isKeyValueSeparated && schema instanceof KeyValueSchema) {
+ KeyValueSchema<?, ?> kvSchema = (KeyValueSchema<?, ?>)
schema;
+ finalSchema =
KeyValueSchemaImpl.of(kvSchema.getKeySchema(), kvSchema.getValueSchema(),
+ KeyValueEncodingType.SEPARATED);
+ } else {
+ finalSchema = schema;
+ }
+
+ sinkRecord = new OutputRecordSinkRecord<>(srcRecord, record) {
Review Comment:
I changed the code to use distinct constructors. It does not introduce a new
class though.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
