apache / apache/pulsar

[Bug] python function ignores `--subs-position`

Open
#17,772 1 comment 0 reactions 0 assignees View on GitHub
Stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Version

Pulsar version: 2.10.1

### Minimal reproduce step

Follow [Pulsar functions quickstart](https://github.com/apache/pulsar/blob/master/site2/website/versioned_docs/version-2.3.0/functions-quickstart.md#writing-and-running-a-new-function). I'm using Docker to run Pulsar standalone:

### 1. Run pulsar standalone

First, create a new Python file:

```bash
$ touch reverse.py
```

In that file, add the following:

```python
def process(input):
return input[::-1]
```

```bash
$ docker run --name pulsar \
--rm -it \
-p 6650:6650 \
-p 8080:8080 \
--volume `pwd`/reverse.py:/pulsar/reverse.py \
apachepulsar/pulsar:2.10.1 bin/pulsar standalone
```

### 2. Produce messages to the source topic

```bash
$ docker exec -it pulsar bash -c "/pulsar/bin/pulsar-client produce \
persistent://public/default/backwards \
--messages '1 tset,2 tset,3 tset'"
```

### 3. Consume results

```bash
$ docker exec -it pulsar bash -c "bin/pulsar-client consume \
persistent://public/default/forwards \
--subscription-name test \
--num-messages 0"
```

### 4. Start the function with --subs-position Earliest

```bash
# using create
$ docker exec -it pulsar bash -c "/pulsar/bin/pulsar-admin functions create \
--py '/pulsar/reverse.py' \
--classname reverse \
--inputs persistent://public/default/backwards \
--output persistent://public/default/forwards \
--tenant public \
--namespace default \
--subs-position Earliest \
--name reverse"

# or localrun
$ docker exec -it pulsar bash -c "/pulsar/bin/pulsar-admin functions localrun \
--py '/pulsar/reverse.py' \
--classname reverse \
--inputs persistent://public/default/backwards \
--output persistent://public/default/forwards \
--tenant public \
--namespace default \
--subs-position Earliest \
--name reverse"
```

### What did you expect to see?

The consumer started in step (3) should output:

```bash
----- got message -----
key:[null], properties:[__pfn_input_msg_id__=CDsQAyAA, __pfn_input_topic__=persistent://public/default/backwards], content:tets 1
----- got message -----
key:[null], properties:[__pfn_input_msg_id__=CDsQBCAA, __pfn_input_topic__=persistent://public/default/backwards], content:tets 2
----- got message -----
key:[null], properties:[__pfn_input_msg_id__=CDsQBSAA, __pfn_input_topic__=persistent://public/default/backwards], content:tets 3
```

### What did you see instead?

Nothing is output by the consumer. Because it starts reading at the latest position.

### Anything else?

In localrun mode you can see the `subscriptionPosition` in the `function_details` is passed to the python function script, but it seems to be ignored:

```bash
$ docker exec -it pulsar bash -c "/pulsar/bin/pulsar-admin functions localrun \
--py '/pulsar/reverse.py' \
--classname reverse \
--inputs persistent://public/default/backwards \
--output persistent://public/default/forwards \
--tenant public \
--namespace default \
--subs-position Earliest \
--name reverse"

2022-09-21T12:09:59,245+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory - Java instance jar location is not defined, using the location defined in system environment : /pulsar/instances/java-instance.jar
2022-09-21T12:09:59,249+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory - Python instance file location is not defined using the location defined in system environment : /pulsar/instances/python-instance/python_instance_main.py
2022-09-21T12:09:59,250+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory - No extra dependencies location is defined in either function worker config or system environment
2022-09-21T12:09:59,328+0000 [main] INFO org.apache.pulsar.functions.runtime.RuntimeSpawner - public/default/reverse-0 RuntimeSpawner starting function
2022-09-21T12:09:59,329+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Creating function log directory /pulsar/logs/functions/public/default/reverse
2022-09-21T12:09:59,329+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Created or found function log directory /pulsar/logs/functions/public/default/reverse
2022-09-21T12:09:59,330+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - ProcessBuilder starting the process with args python /pulsar/instances/python-instance/python_instance_main.py --py /pulsar/reverse.py --logging_directory /pulsar/logs/functions --logging_file reverse --logging_config_file /pulsar/conf/functions-logging/logging_config.ini --instance_id 0 --function_id d094e40d-3b40-4d97-9a48-bfbe3703b3f2 --function_version 29c06a13-31f8-462a-b175-e112a599ae24 --function_details '{"tenant":"public","namespace":"default","name":"reverse","className":"reverse","runtime":"PYTHON","autoAck":true,"parallelism":1,"source":{"inputSpecs":{"persistent://public/default/backwards":{}},"cleanupSubscription":true,"subscriptionPosition":"EARLIEST"},"sink":{"topic":"persistent://public/default/forwards","forwardSourceMessageProperty":true},"resources":{"cpu":1.0,"ram":"1073741824","disk":"10737418240"},"componentType":"FUNCTION"}' --pulsar_serviceurl pulsar://localhost:6650 --use_tls false --tls_allow_insecure false --hostname_verification_enabled false --max_buffered_tuples 1024 --port 45649 --metrics_port 42105 --expected_healthcheck_interval 30 --secrets_provider secretsprovider.ClearTextSecretsProvider --cluster_name local
2022-09-21T12:09:59,340+0000 [main] INFO org.apache.pulsar.functions.runtime.process.ProcessRuntime - Started process successfully
2022-09-21 12:09:59.569 INFO [139811108165440] Client:88 | Subscribing on Topic :persistent://public/default/backwards
2022-09-21 12:09:59.569 INFO [139811108165440] ClientConnection:189 | [ -> pulsar://localhost:6650] Create ClientConnection, timeout=10000
2022-09-21 12:09:59.569 INFO [139811108165440] ConnectionPool:96 | Created connection for pulsar://localhost:6650
2022-09-21 12:09:59.569 INFO [139811045275392] ExecutorService:41 | Run io_service in a single thread
2022-09-21 12:09:59.576 INFO [139811045275392] ClientConnection:375 | [127.0.0.1:33306 -> 127.0.0.1:6650] Connected to broker
2022-09-21 12:09:59.583 INFO [139811045275392] HandlerBase:64 | [persistent://public/default/backwards, public/default/reverse, 0] Getting connection from pool
2022-09-21 12:09:59.585 INFO [139811028408064] ExecutorService:41 | Run io_service in a single thread
2022-09-21 12:09:59.614 INFO [139811045275392] ConsumerImpl:224 | [persistent://public/default/backwards, public/default/reverse, 0] Created consumer on broker [127.0.0.1:33306 -> 127.0.0.1:6650]
2022-09-21 12:10:18.443 INFO [139811045275392] HandlerBase:64 | [persistent://public/default/forwards, ] Getting connection from pool
2022-09-21 12:10:18.452 INFO [139811045275392] ProducerImpl:189 | [persistent://public/default/forwards, ] Created producer on broker [127.0.0.1:33306 -> 127.0.0.1:6650]
```

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!

Contributor guide

Open the contributing guide

Research direction

Reproduce the issue with the Docker standalone commands and inspect the Python runtime entry point at instances/python-instance/python_instance_main.py, especially how the function_details subscriptionPosition is used. Confirm the behavior with messages created before starting the function, then verify that --subs-position Earliest makes those messages reach the output topic and consumer.

Written by the indexing model from the issue text.

Assessment

Tech stack
docker, java, python
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.