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

Code before:
 * limitations under the License.
 */

package com.risingwave.connector;

import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import javax.net.SocketFactory;
import jdk.net.ExtendedSocketOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Custom SocketFactory that enables TCP keep-alive with configurable parameters. This factory
 * configures TCP_KEEPIDLE, TCP_KEEPINTVL, and TCP_KEEPCNT at the socket level.
 */
public class JDBCKeepaliveSocketFactory extends SocketFactory {
    private static final Logger LOG = LoggerFactory.getLogger(JDBCKeepaliveSocketFactory.class);

    private static int keepaliveIdleSeconds = 60;
    private static int keepaliveIntervalSeconds = 10;
    private static int keepaliveCount = 3;

    /** 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

Review comment (suggestion):
Using static mutable fields for configuration creates potential thread-safety issues. If multiple sinks are created concurrently with different TCP keep-alive configurations, they will overwrite each other's settings. Consider using instance fields instead, or implement proper synchronization and per-instance configuration storage.
```suggestion
    private static volatile int keepaliveIdleSeconds = 60;
    private static volatile int keepaliveIntervalSeconds = 10;
    private static volatile int keepaliveCount = 3;

    /** Configure keep-alive parameters. Must be called before creating connections. */
    public static synchronized void configure(int idle, int interval, int count) {
```

Code after:
 * limitations under the License.
 */

package com.risingwave.connector;

import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import javax.net.SocketFactory;
import jdk.net.ExtendedSocketOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Custom SocketFactory that enables TCP keep-alive with configurable parameters. This factory
 * configures TCP_KEEPIDLE, TCP_KEEPINTVL, and TCP_KEEPCNT at the socket level.
 */
public class JDBCKeepaliveSocketFactory extends SocketFactory {
    private static final Logger LOG = LoggerFactory.getLogger(JDBCKeepaliveSocketFactory.class);

    private static class KeepaliveConfig {
        int idleSeconds = 60;
        int intervalSeconds = 10;
        int count = 3;

        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);
    }