Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 15 commits into from
Apr 30, 2024
Merged

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

Show file tree
Hide file tree
Changes from 12 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
35a1797
[Error Message] : Better error message for state mismatch
akashkulk Apr 23, 2024
c4d469a
Update StateManagerFactory.kt
akashkulk Apr 23, 2024
24d14a4
Metadata bumps
akashkulk Apr 23, 2024
f567d37
Merge branch 'master' into akash/error-unable-convert-stream
akashkulk Apr 23, 2024
eaf74ab
Merge branch 'master' into akash/error-unable-convert-stream
akashkulk Apr 23, 2024
ad53ac1
Merge branch 'master' into akash/error-unable-convert-stream
akashkulk Apr 24, 2024
fe75d43
Fix build issues
akashkulk Apr 24, 2024
566dfb7
Merge branch 'master' into akash/error-unable-convert-stream
akashkulk Apr 30, 2024
269bfa6
Bump version
akashkulk Apr 30, 2024
09817cc
Fix tests
akashkulk Apr 30, 2024
c017c13
Merge branch 'master' into akash/error-unable-convert-stream
akashkulk Apr 30, 2024
e260631
Bump cdk
akashkulk Apr 30, 2024
2b549b7
Bump cdk
akashkulk Apr 30, 2024
500e595
Bump cdk
akashkulk Apr 30, 2024
e93041f
Merge branch 'master' into akash/error-unable-convert-stream
akashkulk Apr 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion airbyte-cdk/java/airbyte-cdk/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,8 @@ corresponds to that version.
### Java CDK

| Version | Date | Pull Request | Subject |
| :------ | :--------- | :--------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|:--------|:-----------| :--------------------------------------------------------- |:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
| 0.31.1 | 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 |
Expand Down
2 changes: 1 addition & 1 deletion airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
version=0.31.0
version=0.31.1
9 changes: 5 additions & 4 deletions .../main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactory.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 =
Expand Down Expand Up @@ -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(
Expand Down
5 changes: 3 additions & 2 deletions ...t/kotlin/io/airbyte/cdk/integrations/source/relationaldb/state/StateManagerFactoryTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-mysql/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ plugins {
airbyteJavaConnector {
cdkVersionRequired = '0.30.9'
features = ['db-sources']
useLocalCdk = false
useLocalCdk = true
}

java {
Expand Down
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-mysql/metadata.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 1 addition & 2 deletions .../src/main/java/io/airbyte/integrations/source/mysql/initialsync/MySqlInitialReadUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<>();
Expand Down Expand Up @@ -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,
Expand Down
1 change: 1 addition & 0 deletions docs/integrations/sources/mysql.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. |
Expand Down
Loading