我是独立的 ksql-server 在3节点中与 Kafka 3节点集群。创建了一个 Stream 从 Topic 数据流中有15个分区和数据，可以进行一些扩展。有一段代码 UDF 查找ip2location.bin文件和 UDF 类看起来像：

import java.io.IOException;
import java.util.Map;

import com.google.gson.Gson;

import io.confluent.common.Configurable;
import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;

@UdfDescription(name = "Ip2Lookup", description = "Lookup class for IP2Location database.")
public class Ip2Lookup implements Configurable {

    private IP2Location loc = null;
    private Gson gson = null;

    @Udf(description = "fetches the geoloc of the ipaddress.")
    public synchronized String ip2lookup(String ip) {

        String json = null;
        if (loc != null) {
            IP2LocationResult result = null;
            try {
                result = loc.query(ip);
                System.out.println(result);
                json = gson.toJson(result);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return json;
        }
        return ip;
    }

    @Override
    public void configure(Map<String, ?> arg0) {

        try {
            String db_path = null;
            String os = System.getProperty("os.name").toLowerCase();

            db_path = "/data/md0/ip2loc/ipv4-bin/IP-COUNTRY-REGION-CITY-LATITUDE-LONGITUDE-ZIPCODE-TIMEZONE-ISP-DOMAIN-NETSPEED-AREACODE-WEATHER-MOBILE-ELEVATION-USAGETYPE.BIN";

            loc = new IP2Location(db_path);
            gson = new Gson();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

数据输入到 Topic 然后进入 Stream 相当快（每秒可能有一百万条记录）。与 synchronized 在方法上，速度是每秒3000条记录/消息 ksql-server 节点。以这样的速度，你知道，要赶上这个速度需要时间。没有 synchronized 方法，我看到损坏的数据，因为单个对象/方法被多个线程使用。
问题1：具体情况如何 udf 调用将由ksql调用/调用？
问题2：我可以使用线程处理中的请求吗 udf ?
问题3：作为主题/流是由15个分区组成的，我是否应该旋转15个分区的节点 ksql-servers ?
谢谢。