add keepalive for TCP socket in KubePodProcess

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -7825,7 +7825,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-tiktok-marketing:0.1.4"
+- dockerImage: "airbyte/source-tiktok-marketing:0.1.5"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/tiktok-marketing"
     changelogUrl: "https://docs.airbyte.io/integrations/sources/tiktok-marketing"

airbyte-workers/src/main/java/io/airbyte/workers/process/KubePodProcess.java

@@ -520,6 +520,13 @@ private void setupStdOutAndStdErrListeners() {
       try {
         LOGGER.info("Creating stdout socket server...");
         final var socket = stdoutServerSocket.accept(); // blocks until connected
+        // cat /proc/sys/net/ipv4/tcp_keepalive_time
+        // 300
+        // cat /proc/sys/net/ipv4/tcp_keepalive_probes
+        // 5
+        // cat /proc/sys/net/ipv4/tcp_keepalive_intvl
+        // 60
+        socket.setKeepAlive(true);
         LOGGER.info("Setting stdout...");
         this.stdout = socket.getInputStream();
       } catch (final IOException e) {
@@ -531,6 +538,7 @@ private void setupStdOutAndStdErrListeners() {
       try {
         LOGGER.info("Creating stderr socket server...");
         final var socket = stderrServerSocket.accept(); // blocks until connected
+        socket.setKeepAlive(true);
         LOGGER.info("Setting stderr...");
         this.stderr = socket.getInputStream();
       } catch (final IOException e) {

airbyte-workers/src/main/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteDestination.java

@@ -138,7 +138,7 @@ public boolean isFinished() {
     Preconditions.checkState(destinationProcess != null);
     // As this check is done on every message read, it is important for this operation to be efficient.
     // Short circuit early to avoid checking the underlying process.
-    final var isEmpty = !messageIterator.hasNext();
+    final var isEmpty = !messageIterator.hasNext(); // hasNext is blocking.
     if (!isEmpty) {
       return false;
     }

airbyte-workers/src/main/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteSource.java

@@ -90,7 +90,7 @@ public boolean isFinished() {
     Preconditions.checkState(sourceProcess != null);
     // As this check is done on every message read, it is important for this operation to be efficient.
     // Short circuit early to avoid checking the underlying process.
-    final var isEmpty = !messageIterator.hasNext();
+    final var isEmpty = !messageIterator.hasNext(); // hasNext is blocking.
     if (!isEmpty) {
       return false;
     }