This is an automated email from the ASF dual-hosted git repository.
corgy-w pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 9d6b879051 [Improve][MySQL-CDC] Enhance diagnostics for missing binlog
and GTID during restore (#10566)
9d6b879051 is described below
commit 9d6b87905173e38544f94433b76e475866c7e6ab
Author: Jast <[email protected]>
AuthorDate: Fri May 8 10:34:10 2026 +0800
[Improve][MySQL-CDC] Enhance diagnostics for missing binlog and GTID during
restore (#10566)
---
.../reader/fetch/MySqlSourceFetchTaskContext.java | 313 ++++++++++++++++++++-
.../MysqlCDCWithMissingBinlogDiagnosticsIT.java | 253 +++++++++++++++++
2 files changed, 563 insertions(+), 3 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java
index 0fa0a86065..9e6b03db58 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java
@@ -75,9 +75,12 @@ import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.sql.SQLException;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.offset.BinlogOffset.BINLOG_FILENAME_OFFSET_KEY;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlConnectionUtils.createBinaryClient;
@@ -89,6 +92,8 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlCon
public class MySqlSourceFetchTaskContext extends JdbcSourceFetchTaskContext {
private static final Logger LOG =
LoggerFactory.getLogger(MySqlSourceFetchTaskContext.class);
+ private static final int BINLOG_FILE_SAMPLE_SIZE = 10;
+ private static final Pattern BINLOG_NUMBER_PATTERN =
Pattern.compile("^(.*?)(\\d+)$");
private final MySqlConnection connection;
private final BinaryLogClient binaryLogClient;
@@ -333,10 +338,11 @@ public class MySqlSourceFetchTaskContext extends
JdbcSourceFetchTaskContext {
// And compare with the one we're supposed to use ...
boolean found = logNames.stream().anyMatch(binlogFilename::equals);
if (!found) {
- LOG.info(
- "Connector requires binlog file '{}', but MySQL only has
{}",
+ LOG.warn(
+ "Connector requires binlog file '{}', but it is not
available on the connected MySQL server. {}",
binlogFilename,
- String.join(", ", logNames));
+ summarizeBinlogFiles(logNames));
+ logBinlogNotAvailableDiagnostics(binlogFilename, logNames, offset);
} else {
LOG.info("MySQL has the binlog file '{}' required by the
connector", binlogFilename);
}
@@ -355,6 +361,7 @@ public class MySqlSourceFetchTaskContext extends
JdbcSourceFetchTaskContext {
// Last offsets had GTIDs but the server does not use them ...
LOG.warn(
"Connector used GTIDs previously, but MySQL does not know
of any GTIDs or they are not enabled");
+ logGtidNotAvailableDiagnostics(offset);
return false;
}
@@ -390,14 +397,314 @@ public class MySqlSourceFetchTaskContext extends
JdbcSourceFetchTaskContext {
nonPurgedGtidSetToReplicate);
if (!gtidSetToReplicate.equals(nonPurgedGtidSetToReplicate)) {
LOG.warn("Some of the GTIDs needed to replicate have been
already purged");
+ logGtidNotAvailableDiagnostics(offset);
return false;
}
return true;
}
LOG.info("Connector last known GTIDs are {}, but MySQL has {}",
gtidSet, availableGtidSet);
+ logGtidNotAvailableDiagnostics(offset);
return false;
}
+ private void logBinlogNotAvailableDiagnostics(
+ String requiredBinlogFilename,
+ List<String> availableBinlogFiles,
+ MySqlOffsetContext offset) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: requested starting offset
sourceInfo={}",
+ offset.getSourceInfo());
+ LOG.warn(
+ "MySQL-CDC diagnostic: debezium connection target host={},
port={}",
+ getDbzConnectorConfig().hostname(),
+ getDbzConnectorConfig().port());
+
+ logMySqlServerIdentity();
+ logMySqlBinlogRetentionVariables();
+ logMySqlMasterStatus();
+ logMySqlReplicationStatus();
+ logBinlogRangeAnalysis(requiredBinlogFilename, availableBinlogFiles);
+ }
+
+ private void logGtidNotAvailableDiagnostics(MySqlOffsetContext offset) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: GTID set is not available on server for
restored offset. restored_gtid_set={}, sourceInfo={}",
+ offset.gtidSet(),
+ offset.getSourceInfo());
+ LOG.warn(
+ "MySQL-CDC diagnostic: debezium connection target host={},
port={}",
+ getDbzConnectorConfig().hostname(),
+ getDbzConnectorConfig().port());
+ logMySqlServerIdentity();
+ logMySqlBinlogRetentionVariables();
+ logMySqlMasterStatus();
+ logMySqlReplicationStatus();
+ }
+
+ private void logMySqlServerIdentity() {
+ final String sql =
+ "SELECT @@hostname AS hostname, @@port AS port, @@server_id AS
server_id, "
+ + "@@server_uuid AS server_uuid, @@read_only AS
read_only, "
+ + "@@super_read_only AS super_read_only, @@version AS
version, "
+ + "@@gtid_mode AS gtid_mode";
+ try {
+ connection.query(
+ sql,
+ rs -> {
+ if (!rs.next()) {
+ return;
+ }
+ LOG.warn(
+ "MySQL-CDC diagnostic: connected to MySQL
hostname={}, port={}, server_id={}, server_uuid={}, read_only={},
super_read_only={}, version={}, gtid_mode={}",
+ rs.getString("hostname"),
+ rs.getString("port"),
+ rs.getString("server_id"),
+ rs.getString("server_uuid"),
+ rs.getString("read_only"),
+ rs.getString("super_read_only"),
+ rs.getString("version"),
+ rs.getString("gtid_mode"));
+ });
+ } catch (SQLException e) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: failed to query MySQL server
identity: {}",
+ e.getMessage());
+ }
+ }
+
+ private void logMySqlBinlogRetentionVariables() {
+ logMySqlVariable("log_bin");
+ logMySqlVariable("binlog_format");
+ logMySqlVariable("binlog_expire_logs_seconds");
+ logMySqlVariable("expire_logs_days");
+ }
+
+ private void logMySqlVariable(String variableName) {
+ final String sql = String.format("SHOW VARIABLES LIKE '%s'",
variableName);
+ try {
+ connection.query(
+ sql,
+ rs -> {
+ if (!rs.next()) {
+ LOG.warn("MySQL-CDC diagnostic: variable {} not
found", variableName);
+ return;
+ }
+ LOG.warn(
+ "MySQL-CDC diagnostic: variable {}={}",
+ rs.getString(1),
+ rs.getString(2));
+ });
+ } catch (SQLException e) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: failed to query variable {}: {}",
+ variableName,
+ e.getMessage());
+ }
+ }
+
+ private void logMySqlMasterStatus() {
+ try {
+ connection.query(
+ "SHOW MASTER STATUS",
+ rs -> {
+ if (!rs.next()) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: SHOW MASTER STATUS
returned empty result");
+ return;
+ }
+ String file = safeGetString(rs, "File");
+ String position = safeGetString(rs, "Position");
+ String gtidSet = safeGetString(rs,
"Executed_Gtid_Set");
+ LOG.warn(
+ "MySQL-CDC diagnostic: master status file={},
position={}, executed_gtid_set={}",
+ file,
+ position,
+ gtidSet);
+ });
+ } catch (SQLException e) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: failed to query SHOW MASTER STATUS:
{}", e.getMessage());
+ }
+ }
+
+ private void logMySqlReplicationStatus() {
+ if (logReplicationStatus("SHOW REPLICA STATUS", "replica")) {
+ return;
+ }
+ logReplicationStatus("SHOW SLAVE STATUS", "slave");
+ }
+
+ private boolean logReplicationStatus(String sql, String label) {
+ try {
+ final List<String> row = new ArrayList<>(8);
+ connection.query(
+ sql,
+ rs -> {
+ if (!rs.next()) {
+ return;
+ }
+ String sourceHost =
+ firstNonNull(
+ safeGetString(rs, "Source_Host"),
+ safeGetString(rs, "Master_Host"));
+ String ioRunning =
+ firstNonNull(
+ safeGetString(rs,
"Replica_IO_Running"),
+ safeGetString(rs, "Slave_IO_Running"));
+ String sqlRunning =
+ firstNonNull(
+ safeGetString(rs,
"Replica_SQL_Running"),
+ safeGetString(rs,
"Slave_SQL_Running"));
+ String relayMasterLogFile =
+ firstNonNull(
+ safeGetString(rs,
"Relay_Master_Log_File"),
+ safeGetString(rs, "Master_Log_File"));
+ String execMasterLogPos =
+ firstNonNull(
+ safeGetString(rs,
"Exec_Master_Log_Pos"),
+ safeGetString(rs,
"Read_Master_Log_Pos"));
+ String retrievedGtidSet = safeGetString(rs,
"Retrieved_Gtid_Set");
+ String executedGtidSet = safeGetString(rs,
"Executed_Gtid_Set");
+
+ row.add("source_host=" + sourceHost);
+ row.add("io_running=" + ioRunning);
+ row.add("sql_running=" + sqlRunning);
+ row.add("relay_master_log_file=" + relayMasterLogFile);
+ row.add("exec_master_log_pos=" + execMasterLogPos);
+ if (retrievedGtidSet != null) {
+ row.add("retrieved_gtid_set=" + retrievedGtidSet);
+ }
+ if (executedGtidSet != null) {
+ row.add("executed_gtid_set=" + executedGtidSet);
+ }
+ });
+ if (row.isEmpty()) {
+ LOG.warn("MySQL-CDC diagnostic: {} status empty/unsupported",
label);
+ return true;
+ }
+ LOG.warn("MySQL-CDC diagnostic: {} status {}", label,
String.join(", ", row));
+ return true;
+ } catch (SQLException e) {
+ LOG.warn("MySQL-CDC diagnostic: {} failed: {}", sql,
e.getMessage());
+ return false;
+ }
+ }
+
+ private void logBinlogRangeAnalysis(
+ String requiredBinlogFilename, List<String> availableBinlogFiles) {
+ BinlogFileNumber required =
parseBinlogFileNumber(requiredBinlogFilename);
+ if (required == null) {
+ return;
+ }
+
+ long min = Long.MAX_VALUE;
+ long max = Long.MIN_VALUE;
+ boolean any = false;
+ for (String file : availableBinlogFiles) {
+ BinlogFileNumber parsed = parseBinlogFileNumber(file);
+ if (parsed == null || !required.prefix.equals(parsed.prefix)) {
+ continue;
+ }
+ any = true;
+ min = Math.min(min, parsed.number);
+ max = Math.max(max, parsed.number);
+ }
+ if (!any) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: cannot compare binlog sequence
because available binlog filenames don't match required prefix '{}'",
+ required.prefix);
+ return;
+ }
+
+ if (required.number < min) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: required binlog sequence {}{} is
older than the earliest available {}{}; likely purged/expired binlog on server",
+ required.prefix,
+ required.number,
+ required.prefix,
+ min);
+ } else if (required.number > max) {
+ LOG.warn(
+ "MySQL-CDC diagnostic: required binlog sequence {}{} is
newer than the latest available {}{}; possible connection to an out-of-date
replica, proxy routing to different instance, or RESET MASTER",
+ required.prefix,
+ required.number,
+ required.prefix,
+ max);
+ } else {
+ LOG.warn(
+ "MySQL-CDC diagnostic: required binlog sequence {}{} is
within available range {}{}..{}{} but missing; possible manual binlog deletion,
binlog.index corruption, or connecting to a different MySQL instance
intermittently",
+ required.prefix,
+ required.number,
+ required.prefix,
+ min,
+ required.prefix,
+ max);
+ }
+ }
+
+ private static String summarizeBinlogFiles(List<String> files) {
+ if (files == null || files.isEmpty()) {
+ return "available binlog files: (empty)";
+ }
+ int total = files.size();
+ int headCount = Math.min(BINLOG_FILE_SAMPLE_SIZE, total);
+ int tailCount = Math.min(BINLOG_FILE_SAMPLE_SIZE, Math.max(0, total -
headCount));
+
+ List<String> head = files.subList(0, headCount);
+ List<String> tail =
+ tailCount > 0
+ ? files.subList(total - tailCount, total)
+ : java.util.Collections.emptyList();
+
+ if (tail.isEmpty()) {
+ return String.format(
+ "available binlog files: count=%d, sample=%s", total,
String.join(", ", head));
+ }
+ return String.format(
+ "available binlog files: count=%d, head=[%s], tail=[%s]",
+ total, String.join(", ", head), String.join(", ", tail));
+ }
+
+ private static BinlogFileNumber parseBinlogFileNumber(String filename) {
+ if (filename == null) {
+ return null;
+ }
+ Matcher matcher = BINLOG_NUMBER_PATTERN.matcher(filename);
+ if (!matcher.matches()) {
+ return null;
+ }
+ String prefix = matcher.group(1);
+ String numberStr = matcher.group(2);
+ try {
+ long number = Long.parseLong(numberStr);
+ return new BinlogFileNumber(prefix, number);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ private static String safeGetString(java.sql.ResultSet rs, String
columnLabel) {
+ try {
+ return rs.getString(columnLabel);
+ } catch (SQLException e) {
+ return null;
+ }
+ }
+
+ private static String firstNonNull(String a, String b) {
+ return a != null ? a : b;
+ }
+
+ private static class BinlogFileNumber {
+ private final String prefix;
+ private final long number;
+
+ private BinlogFileNumber(String prefix, long number) {
+ this.prefix = prefix;
+ this.number = number;
+ }
+ }
+
private void validateAndLoadDatabaseHistory(
MySqlOffsetContext offset, MySqlDatabaseSchema schema) {
schema.initializeStorage();
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithMissingBinlogDiagnosticsIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithMissingBinlogDiagnosticsIT.java
new file mode 100644
index 0000000000..7f0929b133
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithMissingBinlogDiagnosticsIT.java
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.mysql;
+
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlContainer;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlVersion;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.UniqueDatabase;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.await;
+
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason = "Currently SPARK and FLINK do not support restore")
+public class MysqlCDCWithMissingBinlogDiagnosticsIT extends TestSuiteBase
implements TestResource {
+ private static final String MYSQL_HOST = "mysql_cdc_e2e";
+ private static final String MYSQL_USER_NAME = "mysqluser";
+ private static final String MYSQL_USER_PASSWORD = "mysqlpw";
+ private static final String MYSQL_DATABASE = "mysql_cdc";
+ private static final String SOURCE_TABLE = "mysql_cdc_e2e_source_table";
+ private static final String SINK_TABLE = "mysql_cdc_e2e_sink_table";
+
+ private static final MySqlContainer MYSQL_CONTAINER =
createMySqlContainer(MySqlVersion.V8_0);
+
+ private final UniqueDatabase inventoryDatabase =
+ new UniqueDatabase(
+ MYSQL_CONTAINER, MYSQL_DATABASE, "mysqluser", "mysqlpw",
MYSQL_DATABASE);
+
+ private static MySqlContainer createMySqlContainer(MySqlVersion version) {
+ return new MySqlContainer(version)
+ .withConfigurationOverride("docker/server-gtids/my.cnf")
+ .withSetupSQL("docker/setup.sql")
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MYSQL_HOST)
+ .withDatabaseName(MYSQL_DATABASE)
+ .withUsername(MYSQL_USER_NAME)
+ .withPassword(MYSQL_USER_PASSWORD)
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
+ }
+
+ private String driverUrl() {
+ return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
+ }
+
+ @TestContainerExtension
+ protected final ContainerExtendedFactory extendedFactory =
+ container -> {
+ Container.ExecResult extraCommands =
+ container.execInContainer(
+ "bash",
+ "-c",
+ "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
+ + driverUrl());
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ };
+
+ @BeforeAll
+ @Override
+ public void startUp() {
+ log.info("Starting Mysql containers...");
+ Startables.deepStart(Stream.of(MYSQL_CONTAINER)).join();
+ log.info("Mysql Containers are started");
+ inventoryDatabase.createAndInitialize();
+ log.info("Mysql ddl execution is complete");
+ }
+
+ @Override
+ @AfterAll
+ public void tearDown() {
+ if (MYSQL_CONTAINER != null) {
+ MYSQL_CONTAINER.close();
+ }
+ }
+
+ @TestTemplate
+ public void
testRestoreFailsAndLogsDiagnosticsWhenBinlogMissing(TestContainer container)
+ throws InterruptedException, IOException {
+ clearTable(MYSQL_DATABASE, SINK_TABLE);
+
+ Long jobId = JobIdGenerator.newJobId();
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.executeJob(
+ "/mysqlcdc_to_mysql_with_binlog_delete.conf",
+ String.valueOf(jobId));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ List<List<Object>> source =
+ query(
+ "select count(1) from "
+ + MYSQL_DATABASE
+ + "."
+ + SOURCE_TABLE);
+ List<List<Object>> sink =
+ query(
+ "select count(1) from "
+ + MYSQL_DATABASE
+ + "."
+ + SINK_TABLE);
+ Assertions.assertEquals(source, sink);
+ });
+
+ // Ensure checkpoint offset is in a newer binlog file
+ executeSql("flush binary logs");
+ executeSql(
+ "INSERT INTO "
+ + MYSQL_DATABASE
+ + "."
+ + SOURCE_TABLE
+ + " (id, f_int, f_varchar) VALUES (200, 200,
'binlog_rotate')");
+
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ query(
+ "select count(1) from "
+ + MYSQL_DATABASE
+ + "."
+ + SOURCE_TABLE),
+ query(
+ "select count(1) from "
+ + MYSQL_DATABASE
+ + "."
+ + SINK_TABLE)));
+
+ Assertions.assertEquals(0,
container.savepointJob(String.valueOf(jobId)).getExitCode());
+
+ // Rotate again, then purge all previous binlogs (including the one
referenced by savepoint)
+ executeSql("flush binary logs");
+ String purgeToBinlog = query("show master
status").get(0).get(0).toString();
+ executeSql("purge binary logs to '" + purgeToBinlog + "'");
+
+ // Restore should fail due to missing binlog, but must output
diagnostics for
+ // troubleshooting
+ Container.ExecResult restoreResult =
+ container.restoreJob(
+ "/mysqlcdc_to_mysql_with_binlog_delete.conf",
String.valueOf(jobId));
+ Assertions.assertNotEquals(
+ 0,
+ restoreResult.getExitCode(),
+ "Expected restore to fail due to missing binlog, but exit code
is 0");
+
+ String logs =
+ restoreResult.getStdout()
+ + "\n"
+ + restoreResult.getStderr()
+ + "\n"
+ + container.getServerLogs();
+ Assertions.assertTrue(
+ logs.contains("MySQL-CDC diagnostic: connected to MySQL"),
+ "Expected diagnostic logs to include MySQL server identity,
but not found in server logs");
+ Assertions.assertTrue(
+ logs.contains("Connector requires binlog file")
+ || logs.contains("MySQL-CDC diagnostic: GTID set is
not available")
+ || logs.contains(
+ "Some of the GTIDs needed to replicate have
been already purged"),
+ "Expected diagnostic logs to include missing binlog or GTID
diagnostics, but not found in logs");
+ }
+
+ private void clearTable(String database, String tableName) {
+ executeSql("truncate table " + database + "." + tableName);
+ }
+
+ private void executeSql(String sql) {
+ try (Connection connection = getJdbcConnection()) {
+ connection.createStatement().execute(sql);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Connection getJdbcConnection() throws SQLException {
+ return DriverManager.getConnection(
+ MYSQL_CONTAINER.getJdbcUrl(),
+ MYSQL_CONTAINER.getUsername(),
+ MYSQL_CONTAINER.getPassword());
+ }
+
+ private List<List<Object>> query(String sql) {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ List<List<Object>> result = new ArrayList<>();
+ int columnCount = resultSet.getMetaData().getColumnCount();
+ while (resultSet.next()) {
+ ArrayList<Object> objects = new ArrayList<>();
+ for (int i = 1; i <= columnCount; i++) {
+ objects.add(resultSet.getObject(i));
+ }
+ result.add(objects);
+ }
+ return result;
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
