apache / apache/pulsar

pulsar functions command: topics-pattern option does not work as intended?

Open
#10,517 2 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**

Try to apply the pulsar function to all the input topic starting with the prefix “test-topic-“.
Based on the description of the option “--topics-pattern” from https://pulsar.apache.org/docs/en/pulsar-admin/, it seems it is possible.
However, it does not work. Is it a functional bug or documentation issue?

**To Reproduce**
Steps to reproduce the behavior:

1. Create simple pulsar function
`
from pulsar import Function
class BasicFunction(Function):
def __init__(self):
pass
def process(self, input, context):
logger = context.get_logger()
logger.info("input {}".format(input))
return input
`

2. Create the pulsar function with pulsar
`
test_topic_out="test-topic-x-out"
test_topic_pattern="test-topic-*"

$pulsar_admin functions \
create \
--tenant public \
--namespace default \
--name BaseFunction \
--py basic_function.py \
--classname basic_function.BasicFunction \
--output $test_topic_out \
--topics-pattern $test_topic_pattern \
--log-topic persistent://public/default/log
`

3. pulsar function logs
`
[2021-05-08 18:58:34 +0000] [INFO] python_instance_main.py: Starting Python instance with Namespace(client_auth_params=None, client_auth_plugin=None, cluster_name='standalone', dependency_repository=None, expected_healthcheck_interval=30, extra_dependency_repository=None, function_details='{"tenant":"public","namespace":"default","name":"BaseFunction","className":"basic_function.BasicFunction","logTopic":"persistent://public/default/log","runtime":"PYTHON","autoAck":true,"parallelism":1,"source":{"inputSpecs":{"test-topic-*":{"isRegexPattern":true}},"cleanupSubscription":true},"sink":{"topic":"test-topic-x-out","forwardSourceMessageProperty":true},"resources":{"cpu":1.0,"ram":"1073741824","disk":"10737418240"},"componentType":"FUNCTION"}', function_id='a6000222-dfd9-4d1e-9f6b-01840f05f951', function_version='f2393128-cdfb-47a7-b550-3d5ae457cdc7', hostname_verification_enabled=None, install_usercode_dependencies=None, instance_id='0', logging_config_file='/pulsar/conf/functions-logging/logging_config.ini', logging_directory='logs//functions', logging_file='BaseFunction', max_buffered_tuples='1024', metrics_port=37303, port=34149, pulsar_serviceurl='pulsar://localhost:6650', py='/pulsar/download/pulsar_functions/public/default/BaseFunction/0/basic_function.py', secrets_provider='secretsprovider.ClearTextSecretsProvider', secrets_provider_config=None, state_storage_serviceurl='bk://127.0.0.1:4181', tls_allow_insecure_connection=None, tls_trust_cert_path=None, use_tls=None)
[2021-05-08 18:58:34 +0000] [INFO] log.py: Setting up producer for log topic persistent://public/default/log
[2021-05-08 18:58:34 +0000] [INFO] server.py: Serving InstanceCommunication on port 34149
`

4. publish message to test-topic-x
`
import pulsar
import sys

publish_topic = 'test-topic-x'
client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer(publish_topic)

for i in range(5):
out = ('hello-pulsar-%d' % i).encode('utf-8')
producer.send(out)
print("sent message {}".format(out))

client.close()
`

5. Nothing when reading from the persistent://public/default/log
`
import pulsar
client = pulsar.Client('pulsar://127.0.0.1:6650')
consumer = client.subscribe(topic='persistent://public/default/log', subscription_name='log', consumer_type=pulsar.ConsumerType.Shared)

while True:
msg = consumer.receive()
try:
print("Received message '{}' id='{}'".format(msg.value(), msg.message_id()))
# Acknowledge successful processing of the message
consumer.acknowledge(msg)
except:
# Message failed to be processed
consumer.negative_acknowledge(msg)

client.close()
`

**Expected behavior**
I will expect there will be message from the persistent://public/default/log

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Desktop (please complete the following information):**
- OS: [e.g. iOS]

**Additional context**
Add any other context about the problem here.

Contributor guide

Open the contributing guide

Research direction

Start by reproducing the issue with the `pulsar-admin functions create` command, using `--topics-pattern test-topic-*`, then publish to `test-topic-x` and inspect the persistent log topic. Determine whether the pattern should match the published topic; done means the documented behavior works or the documentation clearly describes the limitation.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, python
Domain
cli, 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.