apache / apache/pulsar

Can not use Context#Counter in function

Open
#6,589 3 comments 0 reactions 0 assignees View on GitHub
lifecycle/stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 22h
Merged PRs (30d)
142

Description

**Describe the bug**
When I created a demo function(`WordCountFunction`) in the cluster, I got an exception.According to the source code, I found that `Function#Context#stateContext` needs the support of `StreamStorage`.I started `StreamStorage` use `extraServerComponents=org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent` and restart `WordCountFunction`,i found the `WordCountFunction` thread is parking.
In my pulsar cluster,broker and bookie deploy on different machine.
**To Reproduce**
Steps to reproduce the behavior:
1. start `StreamStorage` with a bookie server at machine A
2. start `WordCountFunction` with a broker server at machine B
3. `WordCountFunction` thread is parking at `JavaInstanceRunnable.createStateTable(JavaInstanceRunnable.java:345)`

**Screenshots**
```
19:27:11.455 [test/test-namespace/WordCountFunction-0] INFO org.apache.pulsar.functions.instance.JavaInstanceRunnable - Starting Java Instance WordCountFunction :
Details = tenant: "test"
namespace: "test-namespace"
name: "WordCountFunction"
className: "org.apache.pulsar.functions.api.examples.WordCountFunction"
userConfig: "{\"PublishTopic\":\"test_result\"}"
autoAck: true
parallelism: 1
source {
typeClassName: "java.lang.String"
inputSpecs {
key: "test/test-namespace/test_src"
value {
}
}
cleanupSubscription: true
}
sink {
topic: "test/test-namespace/test_result"
typeClassName: "java.lang.Void"
}
resources {
cpu: 1.0
ram: 1073741824
disk: 10737418240
}
componentType: FUNCTION

19:27:11.455 [test/test-namespace/WordCountFunction-0] INFO org.apache.pulsar.functions.instance.JavaInstanceRunnable - Load JAR: /usr/local/pulsar-2.5.0/download/pulsar_functions/test/test-namespace/WordCountFunction/0/pulsar-functions-api-examples.jar
19:27:11.467 [test/test-namespace/WordCountFunction-0] INFO org.apache.pulsar.functions.instance.JavaInstanceRunnable - Initialize function class loader for function WordCountFunction at function cache manager
19:27:11.920 [client-scheduler-OrderedScheduler-0-0] INFO org.apache.bookkeeper.clients.impl.channel.StorageServerChannelManager - Added range server (hostname: "127.0.0.1"
port: 4181
) into the channel manager.
```
`jstack -l` thread info
```
"test/test-namespace/WordCountFunction-0" #29 prio=5 os_prio=0 tid=0x00007f9215a92800 nid=0x5c06 waiting on condition [0x00007f9144d2d000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00000000efe1e060> (a java.util.concurrent.CompletableFuture$Signaller)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1693)
at java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3323)
at java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1729)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
at org.apache.bookkeeper.common.concurrent.FutureUtils.result(FutureUtils.java:72)
at org.apache.bookkeeper.common.concurrent.FutureUtils.result(FutureUtils.java:61)
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.createStateTable(JavaInstanceRunnable.java:345)
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.setupStateTable(JavaInstanceRunnable.java:397)
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.setupJavaInstance(JavaInstanceRunnable.java:208)
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:244)
at java.lang.Thread.run(Thread.java:748)

Locked ownable synchronizers:
- None
```
**Desktop (please complete the following information):**
- OS: Pulsar 2.5.0
**Additional context**
I find the endpoint registered to zk by `StreamStorage` is `127.0.0.1:4181`,The reason the thread is parked should be this.I started the `WordCountFunction` on the same machine with `StreamStorage`,the function is successful.
I looked at the source code but couldn't find how to modify the configuration registered to zk.

```
ls /stream
[assignment, controller, metadata, servers, storage]
ls /stream/servers
[available]
ls /stream/servers/available
[127.0.0.1:4181]
```

Contributor guide

Open the contributing guide

Research direction

Start at JavaInstanceRunnable.java:345 in createStateTable and trace setupStateTable, then inspect how StreamStorageLifecycleComponent registers its endpoint. Reproduce with StreamStorage and WordCountFunction on separate machines, checking the /stream/servers/available entry in ZooKeeper. Done means the function can create its state table when the advertised storage endpoint is remote rather than 127.0.0.1.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.