Copilot commented on code in PR #4357: URL: https://github.com/apache/flink-cdc/pull/4357#discussion_r3037360912
########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlSnapshotSplitReadTaskConnectionTest.java: ########## @@ -0,0 +1,211 @@ +/* + * 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.flink.cdc.connectors.mysql.debezium.task; + +import org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils; +import org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext; +import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig; +import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory; +import org.apache.flink.cdc.connectors.mysql.table.StartupOptions; + +import com.github.shyiko.mysql.binlog.BinaryLogClient; +import io.debezium.connector.mysql.MySqlConnection; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.ZoneId; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * Tests for {@link MySqlSnapshotSplitReadTask} to verify correct usage of writer and reader + * connections. + * + * <p>The key behavior being tested: + * + * <ul> + * <li>Writer connection is used for binlog position retrieval (SHOW MASTER STATUS) + * <li>Reader connection is used for snapshot data queries (SELECT statements) + * <li>When reader hostname is not configured, reader connection falls back to writer + * </ul> + */ +class MySqlSnapshotSplitReadTaskConnectionTest { + + private static final String PRIMARY_HOSTNAME = "writer-host"; + private static final String SNAPSHOT_HOSTNAME = "reader-host"; + private static final int PORT = 3306; + + @Test + void testStatefulTaskContextWithSeparateConnections() { + MySqlSourceConfig config = createConfig(PRIMARY_HOSTNAME, SNAPSHOT_HOSTNAME); + BinaryLogClient binaryLogClient = mock(BinaryLogClient.class); + + // Create separate writer and reader connections + MySqlConnection writerConnection = DebeziumUtils.createMySqlConnection(config); + MySqlConnection readerConnection = DebeziumUtils.createSnapshotMySqlConnection(config); + + // Create StatefulTaskContext with both connections + StatefulTaskContext context = + new StatefulTaskContext( + config, binaryLogClient, writerConnection, readerConnection); + + // Verify the context returns the correct connections + assertThat(context.getConnection()).isSameAs(writerConnection); + assertThat(context.getSnapshotConnection()).isSameAs(readerConnection); + + // Verify writer and reader connections point to different hosts + assertThat(writerConnection.connectionString()).contains(PRIMARY_HOSTNAME); + assertThat(readerConnection.connectionString()).contains(SNAPSHOT_HOSTNAME); + } + + @Test + void testStatefulTaskContextWithSameConnectionWhenNoReaderConfigured() { + MySqlSourceConfig config = createConfig(PRIMARY_HOSTNAME, null); + BinaryLogClient binaryLogClient = mock(BinaryLogClient.class); + + // When no reader hostname is configured, createReaderMySqlConnection falls back to writer Review Comment: The comment refers to `createReaderMySqlConnection`, but the API being used is `DebeziumUtils.createSnapshotMySqlConnection(...)`. Updating the wording will avoid confusion when maintaining these tests (especially since the production API is consistently named around “snapshot”). ```suggestion // When no reader hostname is configured, createSnapshotMySqlConnection falls back to writer ``` ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java: ########## @@ -73,7 +74,32 @@ public static JdbcConnection openJdbcConnection(MySqlSourceConfig sourceConfig) try { jdbc.connect(); } catch (Exception e) { - LOG.error("Failed to open MySQL connection", e); + LOG.error("Failed to open MySQL connection to {}", sourceConfig.getHostname(), e); + throw new FlinkRuntimeException(e); + } Review Comment: `openJdbcConnection()` now wraps connection failures in `new FlinkRuntimeException(e)` without a message. Since logs may not always be available (or may be rate-limited), consider including the target host/port in the exception message too so callers get actionable context from the thrown error alone (similar for `openSnapshotJdbcConnection`). ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java: ########## @@ -84,6 +85,19 @@ public MySqlSourceConfigFactory hostname(String hostname) { return this; } + /** + * Optional hostname of a MySQL read replica to use for snapshot queries. When set, snapshot + * data will be read from this replica instead of the primary writer instance, reducing the load + * on the writer. The binlog position will still be obtained from the primary writer instance. + * + * <p>An empty string is treated as {@code null}, falling back to the primary writer instance. + */ + public MySqlSourceConfigFactory snapshotHostname(String snapshotHostname) { + this.snapshotHostname = + (snapshotHostname != null && snapshotHostname.trim().isEmpty()) ? null : snapshotHostname; Review Comment: `snapshotHostname()` treats whitespace-only values as absent via `trim().isEmpty()`, but it stores the *untrimmed* hostname when non-empty. This means values like " reader-host " will be accepted yet later used verbatim in JDBC URLs / Debezium config, likely causing connection failures that are hard to diagnose. Consider storing `snapshotHostname.trim()` (or trimming in DebeziumUtils before use) once it’s known to be non-empty. ```suggestion String trimmedSnapshotHostname = snapshotHostname == null ? null : snapshotHostname.trim(); this.snapshotHostname = (trimmedSnapshotHostname == null || trimmedSnapshotHostname.isEmpty()) ? null : trimmedSnapshotHostname; ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
