pulsar IO sink missing after functions worker cluster leader restart

[bohuslav-broza]

hi everyone,
for years we are using Apache Pulsar deployed to K8s and at the moment we are running the version 3.3.5. Lately we extended our Pulsar usage with a Pulsar IO sink. We are dumping data from a topic to an AWS S3 using the cloud storage sink connector version 3.3.5.2 and for pulsar resource management we use pulsar resource operator version 0.9.2.

With our configuration we run into situations, where after restart of the functions worker cluster leader the sink stops operate. Details follows. Any help, hint or suggestion is highly appreciated.

Our cluster consists of two brokers and the functions workers are configured to run on those brokers with kubernetess runtime for the function instance
Functions workers form a cluster and typically broker-1 is the leader

pulsarctl functions-worker get-cluster 
[
  {
    "workerId": "c-pulsar-fw-pulsar-broker-0.pulsar-broker.pulsar.svc.cluster.local-8080",
    "workerHostname": "pulsar-broker-0.pulsar-broker.pulsar.svc.cluster.local",
    "port": 8080
  },
  {
    "workerId": "c-pulsar-fw-pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local-8080",
    "workerHostname": "pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local",
    "port": 8080
  }
]

pulsarctl functions-worker get-cluster-leader
{
  "workerId": "c-pulsar-fw-pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local-8080",
  "workerHostname": "pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local",
  "port": 8080
}

The sink is created using the pulsar resource operator

apiVersion: resource.streamnative.io/v1alpha1
kind: PulsarSink
metadata:
  name: testsink
  namespace: auditlog
spec:
  className: org.apache.pulsar.io.jcloud.sink.CloudStorageGenericRecordSink
  cleanupSubscription: true
  connectionRef:
    name: "pulsar-connection"
  customRuntimeOptions:
    jobNamespace: "auditlog"
    resourceRequirements:
      requests:
        memory: "256Mi"
        cpu: "250m"
      limits:
        memory: "384Mi"
        cpu: "500m"
  inputs:
    - persistent://auditlog/default/testsink
  archive:
    url: builtin://cloud-storage
  tenant: auditlog
  namespace: default
  name: testsink
  sourceSubscriptionName: auditlog-sub-testsink
  parallelism: 1
  sourceSubscriptionPosition: Earliest
  configs:
    provider: "s3v2"
    useDefaultCredentials: "true"
    bucket: "bobrpulsar-testing123"
    region: "us-east-1"
    formatType: "json"
    partitionerType: "PARTITION"
    batchSize: "10000"
    batchTimeMs: "300000"

The sink pod pf-auditlog-default-testsink-0 is created correctly in the auditlog k8s namespce and all the required K8s object - sts and service. The sink is assigned to the broker-0

 pulsarctl functions-worker get-function-assignments
{
  "c-pulsar-fw-pulsar-broker-0.pulsar-broker.pulsar.svc.cluster.local-8080": [
    "auditlog/default/testsink:-1"
  ]
}

When we restart the broker-0 the sink continues to work, however if we restart the leader of the functions worker cluster the broker-0 deletes the sink function (all the K8s objects) after the broker-1 becomes the leader again and we end up in the situation, where there is no pod running with the sink function

pulsarctl functions-worker get-function-assignments
{}

However the operator itself thinks the sink is running fine.

 k get pulsarsink -n auditlog
NAME              RESOURCE_NAME     GENERATION   OBSERVED_GENERATION   READY
testsink          testsink          18           18                    True

Functions worker configuration
broker-1-functions_worker.txt
broker-0-functions_worker.txt

The expected behaviour I require is that during the functions worker cluster changes, which are often in the K8s environment as pods are rescheduled or fail, the sink would survive these events and continue to operate. Any help, hint or suggestion is appreciated.

[lhotari]

Do you find any error logs on the brokers / function workers?

[bohuslav-broza]

hi Lari,
thanks for helping me on this issue! I have reproduced the issue in my environment following way =>

The sink testsink is running. I have restarted 3x broker-0 and 1x broker-1

date; k delete pod -n pulsar pulsar-broker-0
Wed Apr 23 03:13:02 PM CEST 2025
pod "pulsar-broker-0" deleted

date; k delete pod -n pulsar pulsar-broker-0
Wed Apr 23 03:14:56 PM CEST 2025
pod "pulsar-broker-0" deleted


date; k delete pod -n pulsar pulsar-broker-0
Wed Apr 23 03:16:19 PM CEST 2025
pod "pulsar-broker-0" deleted

date; k delete pod -n pulsar pulsar-broker-1
Wed Apr 23 03:18:25 PM CEST 2025
pod "pulsar-broker-1" deleted

After restart of the broker-1 I have ended in a situation where the pipeline is not running but operator sees it in the ready state

root@tiger-tools-756d99c95c-dq6bk:/tools# date; pulsarctl functions-worker get-cluster
Wed Apr 23 01:19:44 PM UTC 2025
[
  {
    "workerId": "c-pulsar-fw-pulsar-broker-0.pulsar-broker.pulsar.svc.cluster.local-8080",
    "workerHostname": "pulsar-broker-0.pulsar-broker.pulsar.svc.cluster.local",
    "port": 8080
  },
  {
    "workerId": "c-pulsar-fw-pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local-8080",
    "workerHostname": "pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local",
    "port": 8080
  }
]root@tiger-tools-756d99c95c-dq6bk:/tools# date; pulsarctl functions-worker get-cluster-leader
Wed Apr 23 01:19:49 PM UTC 2025
{
  "workerId": "c-pulsar-fw-pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local-8080",
  "workerHostname": "pulsar-broker-1.pulsar-broker.pulsar.svc.cluster.local",
  "port": 8080
}root@tiger-tools-756d99c95c-dq6bk:/tools# date; pulsarctl functions-worker get-function-assignments
Wed Apr 23 01:19:54 PM UTC 2025


[bohuslav.broza@mgmt-dev01:~] date; k get pulsarsink -n auditlog
Wed Apr 23 03:20:06 PM CEST 2025
NAME              RESOURCE_NAME     GENERATION   OBSERVED_GENERATION   READY
mia-auditlog001   mia-auditlog001   1            1                     True
testsink          testsink          22           22                    True

Logs from brokers

As for the question, I don't see obvious error in the broker-1 logs, last relevant message to testsink is

2025-04-23T13:19:17,560+0000 [main] INFO  org.apache.pulsar.functions.worker.FunctionRuntimeManager - Received assignment delete: auditlog/default/testsink:-1

in broker-0 logs I see that the worker tries to create internal functions worker topics but this fails as those topics already exist

2025-04-23T13:19:15,348+0000 [pulsar-web-39-4] ERROR org.apache.pulsar.broker.admin.impl.PersistentTopicsBase - [null] Topic persistent://public/functions/assignments already exists
2025-04-23T13:19:15,352+0000 [pulsar-web-39-4] INFO  org.eclipse.jetty.server.RequestLog - 10.253.88.115 - - [23/Apr/2025:13:19:15 +0000] "PUT /admin/v2/persistent/public/functions/assignments?authoritative=false HTTP/1.1" 409 38 "-" "Pulsar-Java-v3.3.5" 6
2025-04-23T13:19:15,450+0000 [pulsar-web-39-7] ERROR org.apache.pulsar.broker.admin.impl.PersistentTopicsBase - [null] Topic persistent://public/functions/coordinate already exists
2025-04-23T13:19:15,451+0000 [pulsar-web-39-7] INFO  org.eclipse.jetty.server.RequestLog - 10.253.88.115 - - [23/Apr/2025:13:19:15 +0000] "PUT /admin/v2/persistent/public/functions/coordinate?authoritative=false HTTP/1.1" 409 38 "-" "Pulsar-Java-v3.3.5" 3
2025-04-23T13:19:15,460+0000 [pulsar-web-39-5] ERROR org.apache.pulsar.broker.admin.impl.PersistentTopicsBase - [null] Topic persistent://public/functions/metadata already exists
2025-04-23T13:19:15,461+0000 [pulsar-web-39-5] INFO  org.eclipse.jetty.server.RequestLog - 10.253.88.115 - - [23/Apr/2025:13:19:15 +0000] "PUT /admin/v2/persistent/public/functions/metadata?authoritative=false HTTP/1.1" 409 38 "-" "Pulsar-Java-v3.3.5" 2