apache/pulsar · error · ValueError

Invalid topicname %s

Error message

Invalid topicname %s

What it means

ContextImpl.ack looks up the consumer registered for the message's topic to acknowledge a message. If the topic is not a known consumer topic and does not even match the '<topic>-partition-N' naming pattern of a partitioned topic, Pulsar raises ValueError('Invalid topicname %s'), meaning ack was called with a topic this instance never subscribed to.

Source

Thrown at pulsar-functions/instance/src/main/python/contextimpl.py:217

        message_conf = {}
      message_conf['properties'] = properties

    if message_conf:
      self.publish_producers[topic_name].send_async(
        output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()), **message_conf)
    else:
      self.publish_producers[topic_name].send_async(
        output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()))

  def ack(self, msgid, topic):
    topic_consumer = None
    if topic in self.consumers:
      topic_consumer = self.consumers[topic]
    else:
      # if this topic is a partitioned topic
      m = re.search(r'(.+)-partition-(\d+)', topic)
      if not m:
        raise ValueError('Invalid topicname %s' % topic)
      elif m.group(1) in self.consumers:
        topic_consumer = self.consumers[m.group(1)]
      else:
        raise ValueError('Invalid topicname %s' % topic)
    topic_consumer.acknowledge(msgid)

  def get_and_reset_metrics(self):
    metrics = self.get_metrics()
    # TODO(sanjeev):- Make this thread safe
    self.reset_metrics()
    return metrics

  def reset_metrics(self):
    # TODO: Make it thread safe
    for user_metric in self.user_metrics_map.values():
      user_metric._sum.set(0.0)
      user_metric._count.set(0.0)

View on GitHub (pinned to 820761864e)

Solutions

1. Ack only messages received from input topics the instance is subscribed to (e.g. use the message object's own topic or context ack semantics).
2. Verify the topic name string exactly matches a topic in the instance's consumer map, including tenant/namespace.
3. If it is a partitioned topic, confirm the name ends with '-partition-<n>' and the base topic has a registered consumer.
4. Use context.get_input_topics() / instance config to confirm subscription names before acking.

Example fix

// before
context.ack('persistent://public/default/my-output-topic', msgid)
// after
context.ack('persistent://public/default/my-input-topic', msgid)  # a topic actually consumed by this instance

consumers = ctx.get_input_topics()  # not a direct API in all versions; verify topic membership
if topic not in known_input_topics and not re.match(r'.+-partition-\d+$', topic):
    raise ValueError(f"cannot ack unknown topic {topic}")

try:
    context.ack(topic, msgid)
except ValueError as e:
    logging.warning('ack rejected: %s', e)  # do not ack unknown topics

When it happens

Trigger: Calling context.ack() with an output/ack topic name that is not in self.consumers and does not contain '-partition-N'; acknowledging a message manually whose topic string was constructed incorrectly.

Common situations: Passing the sink/output topic to ack instead of the input topic; typos in a hardcoded topic name; using an ack on a message obtained from a topic the instance is not consuming.

Related errors

• Invalid short topic name '%s', it should be in the format of
• Invalid topic name: %s
• SourceRecord class type must be PulsarRecord
• Could not import User Function Module %s
• Partitioned topic is not available in effectively_once mode.

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/95f2717b5b20c9bf. Report an issue: GitHub.