apache / apache/pulsar

[improvement] Functions: ability to produce KeyValue<GenericRecord,GenericRecord> output

Open
#14,842 3 comments 0 reactions 1 assignee Claimed by @eolivelli View on GitHub
area/client area/function lifecycle/stale Stale
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

**Is your enhancement request related to a problem? Please describe.**
Currently (Pulsar 2.10) if I code a simply Function like this:

```
@Slf4j
public class MyFunction implements Function> {

@Data
@AllArgsConstructor
public static final class MyKey {
String name;
}
@Data
@AllArgsConstructor
public static final class MyValue {
int age;
}

@Override
public KeyValue apply(GenericObject genericObject) {
log.info("apply to {} {}", genericObject, genericObject.getNativeObject());
return new KeyValue<>(new MyKey("foo"), new MyValue(543));
}
}
```

I am able to successfully consume a KeyValue topic (like a topic written by Debezium or the Kafka Source), but I cannot produce to the output topic, because Pulsar sets the key and value type of the KeyValue to byte[].
It seems to ignore the Generic Type of the KeyVale Output datatype .

This is the error we can see

```
8:00:30.993 [public/default/ciao2-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [public/default/ciao2:0] Uncaught exception in Java Instance
java.lang.ClassCastException: class MyFunction$MyKey cannot be cast to class [B (MyFunction$MyKey is in unnamed module of loader org.apache.pulsar.common.nar.NarClassLoader @70f822e; [B is in module java.base of loader 'bootstrap')
at org.apache.pulsar.client.impl.schema.BytesSchema.encode(BytesSchema.java:28) ~[pulsar-client-original.jar:2.8.0.1.1.29-SNAPSHOT]
at org.apache.pulsar.common.schema.KeyValue.encode(KeyValue.java:107) ~[java-instance.jar:?]
```

**Describe the solution you'd like**
No error happens, out-of-the-box. Data is encoded with KeyValue or KeyValue (and SEPARATE encoding possibly).

**Describe alternatives you've considered**
Setting a schema on the output topic before creating the function (I didn't test this, we should add integration tests for this case as well)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.