docstr: Create a JDBC sink.
  
  :param sql: arbitrary DML query (e.g. insert, update, upsert)
  :param type_info: A RowTypeInfo for query field types.
  :param jdbc_execution_options:  parameters of execution, such as batch size and maximum
                                  retries.
  :param jdbc_connection_options: parameters of connection, such as JDBC URL.
  :return: A JdbcSink.
docstr_tok: ["Create","a","JDBC","sink",".",":","param","sql",":","arbitrary","DML","query","(","e.g",".","insert",",","update",",","upsert",")",":","param","type_info",":","A","RowTypeInfo","for","query","field","types",".",":","param","jdbc_execution_options",":","parameters","of","execution",",","such","as","batch","size","and","maximum","retries",".",":","param","jdbc_connection_options",":","parameters","of","connection",",","such","as","JDBC","URL",".",":","return",":","A","JdbcSink","."]
code: def sink(
      sql: str,
      type_info: RowTypeInfo,
      jdbc_connection_options: "JdbcConnectionOptions",
      jdbc_execution_options: "JdbcExecutionOptions" = None,
  ):
      """
      Create a JDBC sink.
  
      :param sql: arbitrary DML query (e.g. insert, update, upsert)
      :param type_info: A RowTypeInfo for query field types.
      :param jdbc_execution_options:  parameters of execution, such as batch size and maximum
                                      retries.
      :param jdbc_connection_options: parameters of connection, such as JDBC URL.
      :return: A JdbcSink.
      """
      sql_types = []
      gateway = get_gateway()
      JJdbcTypeUtil = (
          gateway.jvm.org.apache.flink.connector.jdbc.utils.JdbcTypeUtil
      )
      for field_type in type_info.get_field_types():
          sql_types.append(
              JJdbcTypeUtil.typeInformationToSqlType(
                  field_type.get_java_type_info()
              )
          )
      j_sql_type = to_jarray(gateway.jvm.int, sql_types)
      output_format_clz = gateway.jvm.Class.forName(
          "org.apache.flink.connector.jdbc.internal.JdbcOutputFormat",
          False,
          get_gateway()
          .jvm.Thread.currentThread()
          .getContextClassLoader(),
      )
      j_int_array_type = to_jarray(gateway.jvm.int, []).getClass()
      j_builder_method = output_format_clz.getDeclaredMethod(
          "createRowJdbcStatementBuilder",
          to_jarray(gateway.jvm.Class, [j_int_array_type]),
      )
      j_builder_method.setAccessible(True)
      j_statement_builder = j_builder_method.invoke(
          None, to_jarray(gateway.jvm.Object, [j_sql_type])
      )
  
      jdbc_execution_options = (
          jdbc_execution_options
          if jdbc_execution_options is not None
          else JdbcExecutionOptions.defaults()
      )
      j_jdbc_sink = (
          gateway.jvm.org.apache.flink.connector.jdbc.JdbcSink.sink(
              sql,
              j_statement_builder,
              jdbc_execution_options._j_jdbc_execution_options,
              jdbc_connection_options._j_jdbc_connection_options,
          )
      )
      return JdbcSink(j_jdbc_sink=j_jdbc_sink)
  
imports: from pyflink.common.typeinfo import RowTypeInfo
  from pyflink.datastream.functions import SinkFunction
  from pyflink.java_gateway import get_gateway
  from pyflink.util.java_utils import to_jarray
label_desc: Use image node_id 2 for calling the JdbcSink obj's underlying member method code with example usage: obj.sink(sql, type_info, jdbc_connection_options, jdbc_execution_options) and returns: JdbcSink