-
Notifications
You must be signed in to change notification settings - Fork 4.1k
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.
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
MySQL CDC sync fails because starting binlog position not found in DB #6425 #9514
Changes from 4 commits
870fe6c
086b446
2b24999
3022590
c40c718
a32eb6a
d006a81
6e98620
abc261a
e5e69c4
a92f70a
2f1b665
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -23,19 +23,23 @@ | |
import io.airbyte.integrations.source.jdbc.AbstractJdbcSource; | ||
import io.airbyte.integrations.source.relationaldb.StateManager; | ||
import io.airbyte.integrations.source.relationaldb.TableInfo; | ||
import io.airbyte.integrations.source.relationaldb.models.CdcState; | ||
import io.airbyte.protocol.models.AirbyteCatalog; | ||
import io.airbyte.protocol.models.AirbyteMessage; | ||
import io.airbyte.protocol.models.AirbyteStream; | ||
import io.airbyte.protocol.models.CommonField; | ||
import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; | ||
import io.airbyte.protocol.models.ConfiguredAirbyteStream; | ||
import io.airbyte.protocol.models.SyncMode; | ||
import java.sql.SQLException; | ||
import java.time.Instant; | ||
import java.util.ArrayList; | ||
import java.util.Iterator; | ||
import java.util.List; | ||
import java.util.Map; | ||
import java.util.Optional; | ||
import java.util.Set; | ||
import java.util.stream.Collectors; | ||
import org.slf4j.Logger; | ||
import org.slf4j.LoggerFactory; | ||
|
||
|
@@ -48,6 +52,7 @@ public class MySqlSource extends AbstractJdbcSource<MysqlType> implements Source | |
public static final String MYSQL_DB_HISTORY = "mysql_db_history"; | ||
public static final String CDC_LOG_FILE = "_ab_cdc_log_file"; | ||
public static final String CDC_LOG_POS = "_ab_cdc_log_pos"; | ||
public static final String CDC_OFFSET = "mysql_cdc_offset"; | ||
public static final List<String> SSL_PARAMETERS = List.of( | ||
"useSSL=true", | ||
"requireSSL=true", | ||
|
@@ -180,8 +185,12 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(final | |
new AirbyteDebeziumHandler(sourceConfig, MySqlCdcTargetPosition.targetPosition(database), MySqlCdcProperties.getDebeziumProperties(), | ||
catalog, true); | ||
|
||
return handler.getIncrementalIterators(new MySqlCdcSavedInfoFetcher(stateManager.getCdcStateManager().getCdcState()), | ||
new MySqlCdcStateHandler(stateManager), new MySqlCdcConnectorMetadataInjector(), emittedAt); | ||
CdcState cdcState = stateManager.getCdcStateManager().getCdcState(); | ||
MySqlCdcSavedInfoFetcher fetcher = new MySqlCdcSavedInfoFetcher(cdcState); | ||
if (cdcState != null) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is it possible to use |
||
checkBinlog(cdcState.getState(), database); | ||
} | ||
return handler.getIncrementalIterators(fetcher, new MySqlCdcStateHandler(stateManager), new MySqlCdcConnectorMetadataInjector(), emittedAt); | ||
} else { | ||
LOGGER.info("using CDC: {}", false); | ||
return super.getIncrementalIterators(database, catalog, tableNameToTable, stateManager, | ||
|
@@ -229,4 +238,42 @@ private CheckedConsumer<JdbcDatabase, Exception> getCheckOperation(String name, | |
}; | ||
} | ||
|
||
private void checkBinlog(JsonNode offset, JdbcDatabase database) { | ||
String binlog = getBinlog(offset); | ||
if (binlog != null && !binlog.isEmpty()) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is it possible to use Optional type here? |
||
if (isBinlogAvailable(binlog, database)) { | ||
LOGGER.info(String.format("Binlog %s is available", binlog)); | ||
} else { | ||
String error = | ||
String.format("Binlog %s is not available. This is a critical error, it means that requested binlog is not present on mysql server. " + | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I believe, we can use Text block here instead of string concatenation https://openjdk.java.net/jeps/378, as we migrate to Java 17. |
||
"To fix data synchronization you need to reset your data. " + | ||
"Please check binlog retention policy configurations.", binlog); | ||
LOGGER.error(error); | ||
throw new RuntimeException(String.format("Binlog %s is not available.", binlog)); | ||
} | ||
} | ||
} | ||
|
||
private boolean isBinlogAvailable(String binlog, JdbcDatabase database) { | ||
try { | ||
List<String> binlogs = database.resultSetQuery(connection -> connection.createStatement().executeQuery("SHOW BINARY LOGS"), | ||
resultSet -> resultSet.getString("Log_name")).collect(Collectors.toList()); | ||
|
||
return !binlog.isEmpty() && binlogs.stream().anyMatch(e -> e.equals(binlog)); | ||
} catch (SQLException e) { | ||
LOGGER.error("Can not get binlog list. Error: ", e); | ||
throw new RuntimeException(e); | ||
} | ||
} | ||
|
||
private String getBinlog(JsonNode offset) { | ||
JsonNode node = offset.get(CDC_OFFSET); | ||
Iterator<Map.Entry<String, JsonNode>> fields = node.fields(); | ||
while (fields.hasNext()) { | ||
Map.Entry<String, JsonNode> jsonField = fields.next(); | ||
return Jsons.deserialize(jsonField.getValue().asText()).path("file").asText(); | ||
} | ||
return null; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why do we return null here? |
||
} | ||
|
||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could
stateManager.getCdcStateManager()
return null?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
no, from what I can see from the code :
public StateManager(final DbState serialized, final ConfiguredAirbyteCatalog catalog) {
this.cdcStateManager = new CdcStateManager(serialized.getCdcState());
this.isCdc = serialized.getCdc();
if (serialized.getCdc() == null) {
this.isCdc = false;
}