code: def test_pulsar_source(self):
      TEST_OPTION_NAME = "pulsar.source.enableAutoAcknowledgeMessage"
      pulsar_source = (
          PulsarSource.builder()
          .set_service_url("pulsar://localhost:6650")
          .set_admin_url("http://localhost:8080")
          .set_topics("ada")
          .set_start_cursor(StartCursor.earliest())
          .set_unbounded_stop_cursor(StopCursor.never())
          .set_bounded_stop_cursor(StopCursor.at_publish_time(22))
          .set_subscription_name("ff")
          .set_consumer_name("test_consumer")
          .set_deserialization_schema(SimpleStringSchema())
          .set_config(TEST_OPTION_NAME, True)
          .set_properties(
              {"pulsar.source.autoCommitCursorInterval": "1000"}
          )
          .build()
      )
  
      ds = self.env.from_source(
          source=pulsar_source,
          watermark_strategy=WatermarkStrategy.for_monotonous_timestamps(),
          source_name="pulsar source",
      )
      ds.print()
      plan = eval(self.env.get_execution_plan())
      self.assertEqual(
          "Source: pulsar source", plan["nodes"][0]["type"]
      )
  
      configuration = get_field_value(
          pulsar_source.get_java_function(), "sourceConfiguration"
      )
      self.assertEqual(
          configuration.getString(
              ConfigOptions.key("pulsar.client.serviceUrl")
              .string_type()
              .no_default_value()
              ._j_config_option
          ),
          "pulsar://localhost:6650",
      )
      self.assertEqual(
          configuration.getString(
              ConfigOptions.key("pulsar.admin.adminUrl")
              .string_type()
              .no_default_value()
              ._j_config_option
          ),
          "http://localhost:8080",
      )
      self.assertEqual(
          configuration.getString(
              ConfigOptions.key("pulsar.consumer.subscriptionName")
              .string_type()
              .no_default_value()
              ._j_config_option
          ),
          "ff",
      )
      test_option = (
          ConfigOptions.key(TEST_OPTION_NAME)
          .boolean_type()
          .no_default_value()
      )
      self.assertEqual(
          configuration.getBoolean(test_option._j_config_option), True
      )
      self.assertEqual(
          configuration.getLong(
              ConfigOptions.key(
                  "pulsar.source.autoCommitCursorInterval"
              )
              .long_type()
              .no_default_value()
              ._j_config_option
          ),
          1000,
      )
  
file_path: flink/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py
imports: from pyflink.common import WatermarkStrategy, SimpleStringSchema, Types, ConfigOptions, Duration
  from pyflink.datastream.connectors import DeliveryGuarantee
  from pyflink.datastream.connectors.pulsar import TopicRoutingMode, MessageDelayer, PulsarSink, PulsarSource, StartCursor, StopCursor, RangeGenerator
  from pyflink.testing.test_case_utils import PyFlinkUTTestCase
  from pyflink.util.java_utils import get_field_value, is_instance_of
label_desc: Use image node_id 1 for calling the FlinkPulsarTest obj's underlying member method code with example usage: obj.test_pulsar_source() without return types