code: def write_to_es6(env):
      ELASTICSEARCH_SQL_CONNECTOR_PATH = "file:///path/to/flink-sql-connector-elasticsearch6-1.16.0.jar"
      env.add_jars(ELASTICSEARCH_SQL_CONNECTOR_PATH)
  
      ds = env.from_collection(
          [{"name": "ada", "id": "1"}, {"name": "luna", "id": "2"}],
          type_info=Types.MAP(Types.STRING(), Types.STRING()),
      )
  
      es_sink = (
          Elasticsearch6SinkBuilder()
          .set_emitter(
              ElasticsearchEmitter.static_index("foo", "id", "bar")
          )
          .set_hosts(["localhost:9200"])
          .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE)
          .set_bulk_flush_max_actions(1)
          .set_bulk_flush_max_size_mb(2)
          .set_bulk_flush_interval(1000)
          .set_bulk_flush_backoff_strategy(
              FlushBackoffType.CONSTANT, 3, 3000
          )
          .set_connection_username("foo")
          .set_connection_password("bar")
          .set_connection_path_prefix("foo-bar")
          .set_connection_request_timeout(30000)
          .set_connection_timeout(31000)
          .set_socket_timeout(32000)
          .build()
      )
  
      ds.sink_to(es_sink).name("es6 sink")
  
      env.execute()
  
file_path: flink/flink-python/pyflink/examples/datastream/connectors/elasticsearch.py
imports: import logging
  import sys
  from pyflink.datastream.connectors.elasticsearch import Elasticsearch6SinkBuilder, Elasticsearch7SinkBuilder, FlushBackoffType, ElasticsearchEmitter
  from pyflink.common import Types
  from pyflink.datastream import StreamExecutionEnvironment
  from pyflink.datastream.connectors import DeliveryGuarantee
label_desc: Use image node_id 1 for calling a global function with example usage: write_to_es6(env) without return types