Can not use Context#Counter in function
- 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
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