This message was deleted.

@Oneeb can you share me the code snippet you are tying with? I can help take a look. Also, we internally uses:

SomeFunction implements Function<GenericObject, Record<GenericObject>>

For some generic processing.

@Neng this is the function Im trying to use:

public class CustomFunction implements Function<GenericRecord, GenericRecord> {
    @Override
    public GenericRecord process(GenericRecord input, Context context) {
        System.out.println("Schema type: " + input.getSchemaType());
        KeyValue<?, ?> kvInput = (KeyValue<?, ?>) input.getNativeObject();
        GenericRecord value = (GenericRecord) kvInput.getValue();
        
        System.out.println("kv Value: " + value);
        System.out.println("GenericRecord Value: " + value.getFields());
        
        String payloadOperation = (String) value.getField("op");
        
        GenericRecord payload = value;
        if (payloadOperation.equals("c") || payloadOperation.equals("r") || payloadOperation.equals("u")) {
            payload = (GenericRecord) value.getField("after");
        } else if (payloadOperation.equals("d")) {
            payload = (GenericRecord) value.getField("before");
        }

        return payload;

    }
}

@Alexander Preuß probably knows

I’m not super familiar with pulsar functions but maybe taking a look at this class might help: https://github.com/apache/pulsar/blob/master/tests/docker-images/java-test-functio[…]/tests/integration/functions/RemoveAvroFieldRecordFunction.java

We used something like:

public class XFunction implements Function<GenericObject, Record<GenericObject>> {
...
}

And then you can

Record<GenericObject> record = (Record<GenericObject>) context.getCurrentRecord();

Schema<T> schema = record.getSchema();

// conversion schema to outputSchema based on your need

FunctionRecordBuilder recordBuilder =
                context.newOutputRecordBuilder(outputSchema).destinationTopic(targetTopic);