Repository: risingwavelabs/risingwave
File: java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JDBCKeepaliveSocketFactory.java (Java)

Code before:
    /** Configure keep-alive parameters. Must be called before creating connections. */
    public static void configure(int idle, int interval, int count) {
        keepaliveIdleSeconds = idle;
        keepaliveIntervalSeconds = interval;
        keepaliveCount = count;
        LOG.info(
                "TCP keep-alive configured: idle={}s, interval={}s, count={}",
                idle,
                interval,
                count);
    }

    @Override
    public Socket createSocket() throws IOException {
        Socket socket = new Socket();
        configureSocket(socket);
        return socket;
    }

    @Override
    public Socket createSocket(String host, int port) throws IOException {
        Socket socket = new Socket(host, port);
        configureSocket(socket);
        return socket;
    }

    @Override
    public Socket createSocket(String host, int port, InetAddress localHost, int localPort)
            throws IOException {
        Socket socket = new Socket(host, port, localHost, localPort);
        configureSocket(socket);
        return socket;
    }

    @Override
    public Socket createSocket(InetAddress host, int port) throws IOException {
        Socket socket = new Socket(host, port);
        configureSocket(socket);
        return socket;
    }

    @Override
    public Socket createSocket(
            InetAddress address, int port, InetAddress localAddress, int localPort)
            throws IOException {
        Socket socket = new Socket(address, port, localAddress, localPort);
        configureSocket(socket);
        return socket;
    }

    private void configureSocket(Socket socket) throws IOException {

Review comment (bug):
The `configure` method doesn't use synchronized access to the static fields, creating a race condition. If multiple threads call this method simultaneously, the values could be partially updated. Consider using `synchronized` keyword or atomic operations to ensure thread-safe configuration updates.

Code after:

        KeepaliveConfig() {}

        KeepaliveConfig(int idle, int interval, int count) {
            this.idleSeconds = idle;
            this.intervalSeconds = interval;
            this.count = count;
        }
    }

    private static final ThreadLocal<KeepaliveConfig> keepaliveConfig =
            ThreadLocal.withInitial(KeepaliveConfig::new);

    /**
     * Configure keep-alive parameters for the current thread. Must be called before creating
     * connections.
     */
    public static void configure(int idle, int interval, int count) {
        keepaliveConfig.set(new KeepaliveConfig(idle, interval, count));
        LOG.info(
                "TCP keep-alive configured for thread {}: idle={}s, interval={}s, count={}",
                Thread.currentThread().getName(),
                idle,
                interval,
                count);
    }

    /** Clear keep-alive configuration for the current thread. */
    public static void clearConfig() {
        keepaliveConfig.remove();
    }

    @Override
    public Socket createSocket() throws IOException {
        Socket socket = new Socket();
        configureSocket(socket);
        return socket;
    }

    @Override
    public Socket createSocket(String host, int port) throws IOException {
        Socket socket = new Socket(host, port);
        configureSocket(socket);
        return socket;
    }

    @Override
    public Socket createSocket(String host, int port, InetAddress localHost, int localPort)
            throws IOException {
        Socket socket = new Socket(host, port, localHost, localPort);
        configureSocket(socket);