Python API: add get_input_topics()

pulsar-client-cpp/python/pulsar/functions/context.py

@@ -146,6 +146,11 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", p
     """
     pass
 
+  @abstractmethod
+  def get_input_topics(self):
+    """Returns the input topics of function"""
+    pass
+
   @abstractmethod
   def get_output_topic(self):
     """Returns the output topic of function"""

pulsar-functions/instance/src/main/python/contextimpl.py

@@ -135,6 +135,9 @@ def record_metric(self, metric_name, metric_value):
 
     self.user_metrics_map[metric_name].observe(metric_value)
 
+  def get_input_topics(self):
+    return list(self.instance_config.function_details.source.inputSpecs.keys())
+
   def get_output_topic(self):
     return self.instance_config.function_details.output
 

site2/docs/functions-api.md

@@ -588,6 +588,7 @@ Method | What it provides
 `get_user_config_map` | Returns the entire user-defined config as a dict
 `record_metric` | Records a per-key [metric](#python-metrics)
 `publish` | Publishes a message to the specified Pulsar topic
+`get_input_topics` | The name(s) of the input(topics), as a list
 `get_output_topic` | The name of the output topic
 `get_output_serde_class_name` | The name of the output [SerDe](#python-serde) class
 `ack` | [Acks](reference-terminology.md#acknowledgment-ack) the message being processed to Pulsar

site2/docs/functions-develop.md

@@ -356,6 +356,8 @@ class ContextImpl(pulsar.Context):
     ...
   def record_metric(self, metric_name, metric_value):
     ...
+  def get_input_topics(self):
+    ...
   def get_output_topic(self):
     ...
   def get_output_serde_class_name(self):