[DB sources] : Add back Debezium heartbeat timeout error (#42055)

Author: akashkulk

airbyte-cdk/java/airbyte-cdk/README.md

@@ -175,6 +175,7 @@ corresponds to that version.
 =======
 | Version     | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:------------|:-----------| :--------------------------------------------------------- |:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.41.7      | 2024-07-17 | [\#42055](https://github.com/airbytehq/airbyte/pull/42055) | Add debezium heartbeat timeout back to shutdown debezium.                                                                |
 | 0.41.6      | 2024-07-17 | [\#41996](https://github.com/airbytehq/airbyte/pull/41996) | Fix java interop compilation issue in Config/TransientErrorException.                                                                |
 | 0.41.5      | 2024-07-16 | [\#42011] (https://github.com/airbytehq/airbyte/pull/42011) | Async consumer accepts null default namespace |
 | 0.41.4      | 2024-07-15 | [\#41959](https://github.com/airbytehq/airbyte/pull/41959) | Allow setting `internal_message` in Config/TransientErrorException. Destinations: shorten error message for INCOMPLETE stream status.                          |

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.41.6
+version=0.41.7

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/DebeziumRecordIterator.kt

@@ -255,16 +255,8 @@ class DebeziumRecordIterator<T>(
     private fun heartbeatPosNotChanging(): Boolean {
         if (this.tsLastHeartbeat == null) {
             return false
-        } else if (!isTest() && receivedFirstRecord) {
-            // Closing debezium due to heartbeat position not changing only exists as an escape
-            // hatch
-            // for testing setups. In production, we rely on the platform heartbeats to kill the
-            // sync
-            // ONLY if we haven't received a record from Debezium. If a record has not been received
-            // from Debezium and the heartbeat isn't changing, the sync should be shut down due to
-            // heartbeat position not changing.
-            return false
         }
+
         val timeElapsedSinceLastHeartbeatTs =
             Duration.between(this.tsLastHeartbeat, LocalDateTime.now())
         return timeElapsedSinceLastHeartbeatTs.compareTo(firstRecordWaitTime) > 0
@@ -288,10 +280,6 @@ class DebeziumRecordIterator<T>(
         }
     }
 
-    private fun isTest(): Boolean {
-        return config.has("is_test") && config["is_test"].asBoolean()
-    }
-
     /**
      * [DebeziumRecordIterator.heartbeatEventSourceField] acts as a cache so that we avoid using
      * reflection to setAccessible for each event

airbyte-integrations/connectors/source-postgres/build.gradle

@@ -12,7 +12,7 @@ java {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.41.6'
+    cdkVersionRequired = '0.41.7'
     features = ['db-sources', 'datastore-postgres']
     useLocalCdk = false
 }

airbyte-integrations/connectors/source-postgres/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 3.5.0
+  dockerImageTag: 3.5.1
   dockerRepository: airbyte/source-postgres
   documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
   githubIssueLabel: source-postgres