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

Code before:
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * 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;

Review comment (suggestion):
I think `static` may lead to concurrency issue. There can be multiple JDBC SINK running on ConnectorNode, and if they do not share the same Keepalive params, they can impact each other. 

Code after:
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * 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,