Use APIs

The following table outlines the APIs that you can use to develop Pulsar Functions in Java, Python, and Go.

Interface	Description	Use case
Language-native interface for Java/Python	No Pulsar-specific libraries or special dependencies required (only core libraries).	Functions that do not require access to the context.
Pulsar Functions SDK for Java/Python/Go	Pulsar-specific libraries that provide a range of functionality not available in the language-native interfaces, such as state management or user configuration.	Functions that require access to the context.
Extended Pulsar Functions SDK for Java	An extension to Pulsar-specific libraries, providing the initialization and close interfaces in Java.	Functions that require initializing and releasing external resources.

Use language-native interface for Java/Python

The language-native interface provides a simple and clean approach to writing Java/Python functions, by adding an exclamation point to all incoming strings and publishing the output string to a topic. It has no external dependencies.

The following examples are language-native functions.

Java

To use a piece of Java code as a "language-native" function, you need to implement the java.util.Function interface. You can include any sort of complex logic inside the apply method to provide more processing capabilities.

import java.util.function.Function;

public class JavaNativeExclamationFunction implements Function<String, String> {
    @Override
    public String apply(String input) {
        return String.format("%s!", input);
    }
}

For more details, see code example.

Python

To use a piece of Python code as a "language-native" function, you must have a method named process as follows. It appends an exclamation point to any string value it receives.

def process(input):
    return "{}!".format(input)

For more details, see code example.

note

Write Pulsar Functions in Python 3. To make sure your functions can run, you need to have Python 3 installed for functions workers and set Python 3 as the default interpreter.

Use SDK for Java/Python/Go

The implementation of Pulsar Functions SDK specifies a functional interface that includes the context object as a parameter.

The following examples use Pulsar Functions SDK for different languages.

Java

When developing a function using the Java SDK, you need to implement the org.apache.pulsar.functions.api.Function interface. It specifies only one method that you need to implement called process.

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;

public class ExclamationFunction implements Function<String, String> {
    @Override
    public String process(String input, Context context) {
        return String.format("%s!", input);
    }
}

For more details, see code example.

The return type of the function can be wrapped in a Record generic which gives you more control over the output messages, such as topics, schemas, properties, and so on. Use the Context::newOutputRecordBuilder method to build this Record output.

import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import org.apache.pulsar.functions.api.Record;

public class RecordFunction implements Function<String, Record<String>> {

    @Override
    public Record<String> process(String input, Context context) throws Exception {
        String output = String.format("%s!", input);
        Map<String, String> properties = new HashMap<>(context.getCurrentRecord().getProperties());
        context.getCurrentRecord().getTopicName().ifPresent(topic -> properties.put("input_topic", topic));

        return context.newOutputRecordBuilder(Schema.STRING)
                .value(output)
                .properties(properties)
                .build();
    }
}

For more details, see code example.

Python

To develop a function using the Python SDK, you need to add the pulsar client dependency to your Python installation.

from pulsar import Function

class ExclamationFunction(Function):
  def __init__(self):
    pass

  def process(self, input, context):
    return input + '!'

For more details, see code example.

Go

To develop a function using the Go SDK, you need to add the pulsar client dependency to your Go installation and provide the name of the function to the pf.Start() method inside the main() method. This registers the function with the Pulsar Functions framework and ensures that the specified function can be invoked when a new message arrives.

package main

import (
	"context"
	"fmt"

	"github.com/apache/pulsar/pulsar-function-go/pf"
)

func HandleRequest(ctx context.Context, in []byte) error{
	fmt.Println(string(in) + "!")
	return nil
}

func main() {
	pf.Start(HandleRequest)
}

For more details, see code example.

Use extended SDK for Java

This extended Pulsar Functions SDK provides two additional interfaces to initialize and release external resources.

• By using the initialize interface, you can initialize external resources which only need one-time initialization when the function instance starts.
• By using the close interface, you can close the referenced external resources when the function instance closes.

note

The extended Pulsar Functions SDK for Java is only available in Pulsar 2.10.0 or later versions. Before using it, you need to set up function workers in Pulsar 2.10.0 or later versions.

The following example uses the extended interface of Pulsar Functions SDK for Java to initialize RedisClient when the function instance starts and release it when the function instance closes.

Java

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import io.lettuce.core.RedisClient;

public class InitializableFunction implements Function<String, String> {
    private RedisClient redisClient;

    private void initRedisClient(Map<String, Object> connectInfo) {
        redisClient = RedisClient.create(connectInfo.get("redisURI"));
    }

    @Override
    public void initialize(Context context) {
        Map<String, Object> connectInfo = context.getUserConfigMap();
        redisClient = initRedisClient(connectInfo);
    }

    @Override
    public String process(String input, Context context) {
        String value = client.get(key);
        return String.format("%s-%s", input, value);
    }

    @Override
    public void close() {
        redisClient.close();
    }
}