diff --git a/airbyte-integrations/connectors/source-mysql/metadata.yaml b/airbyte-integrations/connectors/source-mysql/metadata.yaml index b78f4afd4356..6da06df6a722 100644 --- 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.52.3-gtid-patch-v9 + dockerImageTag: 3.52.3-gtid-patch-v10 dockerRepository: tanishq171/source-mysql documentationUrl: https://docs.airbyte.com/integrations/sources/mysql githubIssueLabel: source-mysql diff --git a/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt b/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt index 85383b146415..85e74d6cb70c 100644 --- a/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt +++ b/airbyte-integrations/connectors/source-mysql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt @@ -279,20 +279,39 @@ class MySqlSourceDebeziumOperations( // // Safety: if information_schema or any SHOW CREATE TABLE call fails, fall back to // existing history rather than partial-replace. Worst-case v7 is no worse than v6. - val rebuiltSchemaHistory: DebeziumSchemaHistory? = - try { - rebuildSchemaHistoryFromSource( - existing = debeziumState.schemaHistory, - referencePosition = savedStateOffset.position, - ) - } catch (e: Exception) { - log.warn(e) { - "Schema history rebuild failed; proceeding with existing history. " + - "${e.message}" + // v10: only rebuild when there is no saved schema history to start from. + // + // The v7 full-replace above breaks in-place schema evolution. When a column is added on + // the source, the SHOW CREATE TABLE rebuild seeds the connector with the *current* + // (post-ALTER) table shape, but the connector resumes streaming from a saved binlog + // offset that predates the ALTER. Debezium then validates old, pre-ALTER row events + // (N-1 columns) against the rebuilt N-column schema and aborts with + // "internal schema size N, but row size N-1, restart connector with schema recovery + // mode". Stock Debezium instead replays the ALTER DDL from the binlog at its correct + // position and evolves the schema in lockstep — which is how a new column is normally + // auto-propagated. + // + // So: keep the saved schema history whenever it is non-empty (let binlog DDL replay + // drive schema evolution), and fall back to the SHOW CREATE TABLE rebuild only when + // there is no saved history to resume from — the empty-history case the v7 rebuild was + // originally meant to cover. (A non-empty-but-incomplete history — e.g. a table missing + // after a swap + retention gap — is no longer force-rebuilt; that trade is accepted here + // in favour of not breaking column additions.) + val schemaHistory: DebeziumSchemaHistory? = + debeziumState.schemaHistory?.takeIf { it.wrapped.isNotEmpty() } + ?: try { + rebuildSchemaHistoryFromSource( + existing = debeziumState.schemaHistory, + referencePosition = savedStateOffset.position, + ) + } catch (e: Exception) { + log.warn(e) { + "Schema history rebuild failed; proceeding with existing history. " + + "${e.message}" + } + debeziumState.schemaHistory } - debeziumState.schemaHistory - } - return ValidDebeziumWarmStartState(debeziumState.offset, rebuiltSchemaHistory) + return ValidDebeziumWarmStartState(debeziumState.offset, schemaHistory) } /**