[DB source error messages] : Better error messages when switching between global/per-stream modes (#37507)

akashkulk · web-flow · commit 2b260de9d077 · 2024-04-30T19:18:43.000-04:00
diff --git a/airbyte-cdk/java/airbyte-cdk/README.md b/airbyte-cdk/java/airbyte-cdk/README.md
@@ -173,7 +173,8 @@ corresponds to that version.
 ### Java CDK
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
-| :------ | :--------- | :--------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+|:--------|:-----------| :--------------------------------------------------------- |:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.31.2  | 2024-04-30 | [\#37507](https://github.com/airbytehq/airbyte/pull/37507) | Better error messages when switching between global/per-stream modes.                                                                                          |
 | 0.31.0  | 2024-04-26 | [\#37584](https://github.com/airbytehq/airbyte/pull/37584) | Update S3 destination deps to exclude zookeeper and hadoop-yarn-common                                                                                         |
 | 0.30.11 | 2024-04-25 | [\#36899](https://github.com/airbytehq/airbyte/pull/36899) | changes for bigQuery destination.                                                                                                                              |
 | 0.30.10 | 2024-04-24 | [\#37541](https://github.com/airbytehq/airbyte/pull/37541) | remove excessive logging                                                                                                                                       |
diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties b/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties
@@ -1 +1 @@
-version=0.31.0
+version=0.31.2
diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactory.kt b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactory.kt
@@ -4,6 +4,7 @@
 package io.airbyte.cdk.integrations.source.relationaldb.state
 
 import io.airbyte.cdk.integrations.source.relationaldb.models.DbState
+import io.airbyte.commons.exceptions.ConfigErrorException
 import io.airbyte.commons.json.Jsons
 import io.airbyte.protocol.models.v0.AirbyteStateMessage
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
@@ -95,8 +96,8 @@ object StateManagerFactory {
 
         when (airbyteStateMessage.type) {
             AirbyteStateMessage.AirbyteStateType.STREAM ->
-                throw IllegalArgumentException(
-                    "Unable to convert connector state from stream to global.  Please reset the connection to continue."
+                throw ConfigErrorException(
+                    "You've changed replication modes - please reset the streams in this connector"
                 )
             AirbyteStateMessage.AirbyteStateType.LEGACY -> {
                 globalStateMessage =
@@ -127,8 +128,8 @@ object StateManagerFactory {
         val streamStates: MutableList<AirbyteStateMessage> = ArrayList()
         when (airbyteStateMessage.type) {
             AirbyteStateMessage.AirbyteStateType.GLOBAL ->
-                throw IllegalArgumentException(
-                    "Unable to convert connector state from global to stream.  Please reset the connection to continue."
+                throw ConfigErrorException(
+                    "You've changed replication modes - please reset the streams in this connector"
                 )
             AirbyteStateMessage.AirbyteStateType.LEGACY ->
                 streamStates.addAll(
diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/test/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactoryTest.kt b/airbyte-cdk/java/airbyte-cdk/db-sources/src/test/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactoryTest.kt
@@ -6,6 +6,7 @@ package io.airbyte.cdk.integrations.source.relationaldb.state
 import io.airbyte.cdk.integrations.source.relationaldb.models.CdcState
 import io.airbyte.cdk.integrations.source.relationaldb.models.DbState
 import io.airbyte.cdk.integrations.source.relationaldb.models.DbStreamState
+import io.airbyte.commons.exceptions.ConfigErrorException
 import io.airbyte.commons.json.Jsons
 import io.airbyte.protocol.models.v0.*
 import java.util.List
@@ -160,7 +161,7 @@ class StateManagerFactoryTest {
                         .withStreamState(Jsons.jsonNode(DbStreamState()))
                 )
 
-        Assertions.assertThrows(IllegalArgumentException::class.java) {
+        Assertions.assertThrows(ConfigErrorException::class.java) {
             StateManagerFactory.createStateManager(
                 AirbyteStateMessage.AirbyteStateType.GLOBAL,
                 List.of(airbyteStateMessage),
@@ -280,7 +281,7 @@ class StateManagerFactoryTest {
                 .withType(AirbyteStateMessage.AirbyteStateType.GLOBAL)
                 .withGlobal(globalState)
 
-        Assertions.assertThrows(IllegalArgumentException::class.java) {
+        Assertions.assertThrows(ConfigErrorException::class.java) {
             StateManagerFactory.createStateManager(
                 AirbyteStateMessage.AirbyteStateType.STREAM,
                 List.of(airbyteStateMessage),
diff --git a/airbyte-integrations/connectors/source-mysql/build.gradle b/airbyte-integrations/connectors/source-mysql/build.gradle
@@ -6,7 +6,7 @@ plugins {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.30.9'
+    cdkVersionRequired = '0.31.2'
     features = ['db-sources']
     useLocalCdk = false
 }
diff --git a/airbyte-integrations/connectors/source-mysql/metadata.yaml b/airbyte-integrations/connectors/source-mysql/metadata.yaml
@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 3.3.22
+  dockerImageTag: 3.3.23
   dockerRepository: airbyte/source-mysql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
   githubIssueLabel: source-mysql
diff --git a/airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/initialsync/MySqlInitialReadUtil.java b/airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/initialsync/MySqlInitialReadUtil.java
@@ -89,7 +89,6 @@ public static List<AutoCloseableIterator<AirbyteMessage>> getCdcReadIterators(fi
                                                                                 final String quoteString) {
     final JsonNode sourceConfig = database.getSourceConfig();
     final Duration firstRecordWaitTime = RecordWaitTimeUtil.getFirstRecordWaitTime(sourceConfig);
-    final Duration subsequentRecordWaitTime = RecordWaitTimeUtil.getSubsequentRecordWaitTime(sourceConfig);
     LOGGER.info("First record waiting time: {} seconds", firstRecordWaitTime.getSeconds());
     // Determine the streams that need to be loaded via primary key sync.
     final List<AutoCloseableIterator<AirbyteMessage>> initialLoadIterator = new ArrayList<>();
@@ -160,7 +159,7 @@ public static List<AutoCloseableIterator<AirbyteMessage>> getCdcReadIterators(fi
     }
 
     // Build the incremental CDC iterators.
-    final AirbyteDebeziumHandler<MySqlCdcPosition> handler = new AirbyteDebeziumHandler<>(
+    final AirbyteDebeziumHandler<MySqlCdcPosition> handler = new AirbyteDebeziumHandler<MySqlCdcPosition>(
         sourceConfig,
         MySqlCdcTargetPosition.targetPosition(database),
         true,
diff --git a/docs/integrations/sources/mysql.md b/docs/integrations/sources/mysql.md
@@ -223,6 +223,7 @@ Any database or table encoding combination of charset and collation is supported
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                         |
 |:--------|:-----------|:-----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.3.23  | 2024-04-23 | [37507](https://github.com/airbytehq/airbyte/pull/37507)                                                          | Better errors when user switches from CDC to non-CDC mode.                                                                                           |
 | 3.3.22  | 2024-04-22 | [37541](https://github.com/airbytehq/airbyte/pull/37541)   | Adopt latest CDK. reduce excessive logs.                                                                                                        |
 | 3.3.21  | 2024-04-22 | [37476](https://github.com/airbytehq/airbyte/pull/37476)   | Adopt latest CDK.                                                                                                                               |
 | 3.3.20  | 2024-04-16 | [37111](https://github.com/airbytehq/airbyte/pull/37111)   | Populate null values in record message.                                                                                                         |

Original file line number	Diff line number	Diff line change
`@@ -6,7 +6,7 @@ plugins {`
`6`	`6`	`}`
`7`	`7`
`8`	`8`	`airbyteJavaConnector {`
`9`		`- cdkVersionRequired = '0.30.9'`
	`9`	`+ cdkVersionRequired = '0.31.2'`
`10`	`10`	`features = ['db-sources']`
`11`	`11`	`useLocalCdk = false`
`12`	`12`	`}`