code: def test_pulsar_sink(self):
      ds = self.env.from_collection(
          [("ab", 1), ("bdc", 2), ("cfgs", 3), ("deeefg", 4)],
          type_info=Types.ROW([Types.STRING(), Types.INT()]),
      )
  
      TEST_OPTION_NAME = "pulsar.producer.chunkingEnabled"
      pulsar_sink = (
          PulsarSink.builder()
          .set_service_url("pulsar://localhost:6650")
          .set_admin_url("http://localhost:8080")
          .set_producer_name("fo")
          .set_topics("ada")
          .set_serialization_schema(SimpleStringSchema())
          .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE)
          .set_topic_routing_mode(TopicRoutingMode.ROUND_ROBIN)
          .delay_sending_message(
              MessageDelayer.fixed(Duration.of_seconds(12))
          )
          .set_config(TEST_OPTION_NAME, True)
          .set_properties(
              {"pulsar.producer.batchingMaxMessages": "100"}
          )
          .build()
      )
  
      ds.sink_to(pulsar_sink).name("pulsar sink")
  
      plan = eval(self.env.get_execution_plan())
      self.assertEqual("pulsar sink: Writer", plan["nodes"][1]["type"])
      configuration = get_field_value(
          pulsar_sink.get_java_function(), "sinkConfiguration"
      )
      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.producer.producerName")
              .string_type()
              .no_default_value()
              ._j_config_option
          ),
          "fo - %s",
      )
  
      j_pulsar_serialization_schema = get_field_value(
          pulsar_sink.get_java_function(), "serializationSchema"
      )
      j_serialization_schema = get_field_value(
          j_pulsar_serialization_schema, "serializationSchema"
      )
      self.assertTrue(
          is_instance_of(
              j_serialization_schema,
              "org.apache.flink.api.common.serialization.SimpleStringSchema",
          )
      )
  
      self.assertEqual(
          configuration.getString(
              ConfigOptions.key("pulsar.sink.deliveryGuarantee")
              .string_type()
              .no_default_value()
              ._j_config_option
          ),
          "at-least-once",
      )
  
      j_topic_router = get_field_value(
          pulsar_sink.get_java_function(), "topicRouter"
      )
      self.assertTrue(
          is_instance_of(
              j_topic_router,
              "org.apache.flink.connector.pulsar.sink.writer.router.RoundRobinTopicRouter",
          )
      )
  
      j_message_delayer = get_field_value(
          pulsar_sink.get_java_function(), "messageDelayer"
      )
      delay_duration = get_field_value(
          j_message_delayer, "delayDuration"
      )
      self.assertEqual(delay_duration, 12000)
  
      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.producer.batchingMaxMessages")
              .long_type()
              .no_default_value()
              ._j_config_option
          ),
          100,
      )
  
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 9 for calling the FlinkPulsarTest obj's underlying member method code with example usage: obj.test_pulsar_sink() without return types