Manage Functions

StreamNative Cloud enables you to manage Pulsar Functions by using a variety of tools, including snctl, pulsarctl, pulsar-admin, REST API, and Terraform.

If you want to update or delete functions using snctl, pulsarctl or pulsar-admin, make sure you have set up your client tool. For more information, see set up client tools.

Update a function

When you want to modify configurations or update resources for functions, you can update functions using multiple tools. The following example shows how to update the parallelism of the Java exclamation function exclamation to 2 using different tools.

snctl
pulsarctl
pulsar-admin
Terraform
REST API

snctl pulsar admin functions update \
--name exclamation \
--parallelism 2

You should see the following output:

Updated successfully

And you can further check the status:

snctl pulsar admin functions status --name exclamation

{
  "numInstances": 2,
  "numRunning": 2,
  "instances": [
    {
      "instanceId": 0,
      "status": {
        "running": true,
        "error": "",
        "numRestarts": 0,
        "numReceivedFromSource": 1799,
        "numSystemExceptions": 0,
        "latestSystemExceptions": [],
        "numSourceExceptions": 0,
        "latestSourceExceptions": [],
        "numWritten": 1799,
        "lastReceivedTime": 1693946327331,
        "workerId": "test"
      }
    },
    {
      "instanceId": 1,
      "status": {
        "running": true,
        "error": "",
        "numRestarts": 0,
        "numReceivedFromSource": 689,
        "numSystemExceptions": 0,
        "latestSystemExceptions": [],
        "numSourceExceptions": 0,
        "latestSourceExceptions": [],
        "numWritten": 689,
        "lastReceivedTime": 1693946327129,
        "workerId": "test"
      }
    }
  ]
}

pulsarctl functions update \
--name exclamation \
--parallelism 2

You should see the following output:

Updated successfully

And you can further check the status:

pulsarctl functions status --name exclamation

{
  "numInstances": 2,
  "numRunning": 2,
  "instances": [
    {
      "instanceId": 0,
      "status": {
        "running": true,
        "error": "",
        "numRestarts": 0,
        "numReceivedFromSource": 1799,
        "numSystemExceptions": 0,
        "latestSystemExceptions": [],
        "numSourceExceptions": 0,
        "latestSourceExceptions": [],
        "numWritten": 1799,
        "lastReceivedTime": 1693946327331,
        "workerId": "test"
      }
    },
    {
      "instanceId": 1,
      "status": {
        "running": true,
        "error": "",
        "numRestarts": 0,
        "numReceivedFromSource": 689,
        "numSystemExceptions": 0,
        "latestSystemExceptions": [],
        "numSourceExceptions": 0,
        "latestSourceExceptions": [],
        "numWritten": 689,
        "lastReceivedTime": 1693946327129,
        "workerId": "test"
      }
    }
  ]
}

pulsar-admin \
    --admin-url "${WEB_SERVICE_URL}" \
    --auth-plugin org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2 \
    --auth-params '{"privateKey":"file:///YOUR-KEY-FILE-PATH",
        "issuerUrl":"https://auth.streamnative.cloud/",
        "audience":"urn:sn:pulsar:${orgName}:${instanceName}}'
functions update \
--name exclamation \
--parallelism 2

Replace the placeholder variables with the actual values, which you can get when you set up client tools.

admin-url: the HTTP service URL of your Pulsar cluster.
private_key: the path to the downloaded OAuth2 key file.
audience: the Uniform Resource Name (URN), which is a combination of the urn:sn:pulsar, your organization name, and your Pulsar instance name.
- ${orgName}: the name of your organization.
- ${instanceName}: the name of your instance.

You should see the following output:

Updated successfully

And you can further check the status:

pulsar-admin functions status --name test
{
  "numInstances" : 2,
  "numRunning" : 2,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : true,
      "error" : "",
      "numRestarts" : 0,
      "numReceivedFromSource" : 1287,
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "numSourceExceptions" : 0,
      "latestSourceExceptions" : [ ],
      "numWritten" : 1287,
      "lastReceivedTime" : 1693946605095,
      "workerId" : "test"
    }
  }, {
    "instanceId" : 1,
    "status" : {
      "running" : true,
      "error" : "",
      "numRestarts" : 0,
      "numReceivedFromSource" : 909,
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "numSourceExceptions" : 0,
      "latestSourceExceptions" : [ ],
      "numWritten" : 909,
      "lastReceivedTime" : 1693946605023,
      "workerId" : "test"
    }
  } ]
}

To update the submitted function, you only need to update the Terraform file and then call the following command.

terraform apply

You should see something like following:

pulsar_function.function-1: Refreshing state... [id=public/default/function1]

Terraform used the selected providers to generate the following execution plan. Resource actions are indicated with the following symbols:
  ~ update in-place

Terraform will perform the following actions:

  # pulsar_function.function-1 will be updated in-place
  ~ resource "pulsar_function" "function-1" {
      ~ custom_runtime_options          = jsonencode(
          ~ {
              - clusterName         = "oxia-test"
              - enableStateStore    = false
              - inputTypeClassName  = "java.lang.String"
              - logTopicAgent       = "runtime"
              - managed             = true
              - maxReplicas         = 0
              - outputTypeClassName = "java.lang.String"
              - runnerImage         = "streamnative/pulsar-functions-sn-java-runner:3.1.0.4"
              - serviceAccountName  = "oxia-test-function-pulsarcluster"
                # (1 unchanged attribute hidden)
            }
        )
      ~ disk_mb                         = 10240 -> 128
        id                              = "public/default/function1"
        name                            = "function1"
      ~ parallelism                     = 1 -> 2
      ~ ram_mb                          = 140 -> 128
        # (21 unchanged attributes hidden)
    }

Plan: 0 to add, 1 to change, 0 to destroy.

Do you want to perform these actions?
  Terraform will perform the actions described above.
  Only 'yes' will be accepted to approve.

  Enter a value:

Enter yes and then you should get:

pulsar_function.function-1: Modifying... [id=public/default/function1]
pulsar_function.function-1: Modifications complete after 0s [id=public/default/function1]

Apply complete! Resources: 0 added, 1 changed, 0 destroyed.

For the REST API you can simply call CURL to update your function.

  curl -X POST ${WEB_SERVICE_URL}/admin/v3/functions/public/default/${FUNCTION_NAME} \
  -H 'Authorization: Bearer ${TOKEN}' \
  -H "Content-Type: multipart/form-data" \
  -F 'functionConfig={"name": "${FUNCTION_NAME}", "parallelism": 2};type=application/json' \
  -F 'url=function://public/default/${FUNCTION_NAME}@${FUNCTION_VERSION}' \

Replace the placeholder variables with the actual values that you can get when setting up client tools.

WEB_SERVICE_URL: the HTTPS service URL of your Pulsar cluster.
TOKEN: a valid token to interact with your Pulsar cluster.
FUNCTION_NAME: the name of your function.
FUNCTION_VERSION: the version of your function you want to deploy e.g. 1.12.

Delete a function

The following example shows how to delete the Java exclamation function exclamation using different tools.

snctl
pulsarctl
pulsar-admin
Terraform

To delete the function exclamation, use the following command.

snctl pulsar admin functions delete --tenant public --namespace default --name exclamation

You should see the following output:

Deleted exclamation successfully

If you want to verify whether the function has been deleted successfully, run the following command.

snctl pulsar admin functions get --tenant public --namespace default --name exclamation

You should see the following output:

[✖]  code: 500 reason: failed to perform the request: responseCode: 404, responseMessage: functions.compute.functionmesh.io "exclamation-XXXXX" not found

To delete the function exclamation, use the following command.

pulsarctl functions delete --tenant public --namespace default --name exclamation

You should see the following output:

Deleted exclamation successfully

If you want to verify whether the function has been deleted successfully, run the following command.

pulsarctl functions get --tenant public --namespace default --name exclamation

You should see the following output:

[✖]  code: 500 reason: failed to perform the request: responseCode: 404, responseMessage: functions.compute.functionmesh.io "exclamation-XXXXX" not found

To delete the function exclamation, run the following command.

./bin/pulsar-admin functions delete  --tenant public --namespace default --name exclamation

You should see the following output:

Delete exclamation successfully

To verify the function has been deleted, run the following command.

./bin/pulsar-admin functions get --tenant public --namespace default --name exclamation

You should see the following output:

failed to perform the request: responseCode: 404, responseMessage: functions.compute.functionmesh.io "exclamation-XXXXX" not found

To delete the function exclamation with terraform, run the following command and type yes on the prompt.

terraform destroy

You should see the following output:

pulsar_function.function-1: Refreshing state... [id=public/default/function1]

Terraform used the selected providers to generate the following execution plan. Resource actions are indicated with the following symbols:
  - destroy

Terraform will perform the following actions:

  # pulsar_function.function-1 will be destroyed
  - resource "pulsar_function" "function-1" {
      - auto_ack                        = true -> null
      - classname                       = "org.apache.pulsar.functions.api.examples.ExclamationFunction" -> null
      - cleanup_subscription            = true -> null
      - cpu                             = 0.5 -> null
      - custom_runtime_options          = jsonencode(
            {
              - clusterName         = "oxia-test"
              - enableStateStore    = false
              - env                 = {
                  - HELLO = "WORLD"
                }
              - inputTypeClassName  = "java.lang.String"
              - logTopicAgent       = "runtime"
              - managed             = true
              - maxReplicas         = 0
              - outputTypeClassName = "java.lang.String"
              - runnerImage         = "streamnative/pulsar-functions-sn-java-runner:3.1.0.4"
              - serviceAccountName  = "oxia-test-function-pulsarcluster"
            }
        ) -> null
      - dead_letter_topic               = "public/default/dlt" -> null
      - disk_mb                         = 10240 -> null
      - forward_source_message_property = true -> null
      - id                              = "public/default/function1" -> null
      - inputs                          = [
          - "persistent://public/default/test-java-input",
        ] -> null
      - jar                             = "function://public/default/exclamation@v0.1" -> null
      - log_topic                       = "public/default/lt" -> null
      - max_message_retries             = 100 -> null
      - name                            = "function1" -> null
      - namespace                       = "default" -> null
      - output                          = "persistent://public/default/test-java-output" -> null
      - parallelism                     = 2 -> null
      - processing_guarantees           = "ATLEAST_ONCE" -> null
      - ram_mb                          = 140 -> null
      - retain_key_ordering             = true -> null
      - retain_ordering                 = false -> null
      - secrets                         = jsonencode(
            {
              - SECRET1 = {
                  - key  = "hello"
                  - path = "sectest"
                }
            }
        ) -> null
      - skip_to_latest                  = true -> null
      - subscription_name               = "test-sub" -> null
      - subscription_position           = "Latest" -> null
      - tenant                          = "public" -> null
      - timeout_ms                      = 6666 -> null
    }

Plan: 0 to add, 0 to change, 1 to destroy.

Do you really want to destroy all resources?
  Terraform will destroy all your managed infrastructure, as shown above.
  There is no undo. Only 'yes' will be accepted to confirm.

  Enter a value:

Enter yes and you will get:

pulsar_function.function-1: Destroying... [id=public/default/function1]
pulsar_function.function-1: Destruction complete after 0s

Destroy complete! Resources: 1 destroyed.

What’s next?

Learn how to configure stateful functions.
Learn how to monitor functions.
Reference common configurations.

Get Started

Clusters

Data Streams

Process

Connect

Lakehouse

Governance

Pulsar Clients

MQTT Clients

Tools

Pulsar Changelogs

Update a function

Delete a function

What’s next?

​Update a function

​Delete a function

​What’s next?

Update a function

Delete a function

What’s next?