[improvement] Functions: ability to produce KeyValue<GenericRecord,GenericRecord> output
- 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
Assessment
This issue has not been assessed yet.