fix: cache showBinaryLogStatement in each MySqlConnection object

Signed-off-by: yuxiqian <34335406+yuxiqian@users.noreply.github.com>

Author: yuxiqian

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlConnection.java

@@ -76,8 +76,7 @@ public class MySqlConnection extends JdbcConnection {
 
     private static String MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT = "SHOW MASTER STATUS";
     private static String MYSQL_NEW_SHOW_BINARY_LOG_STATEMENT = "SHOW BINARY LOG STATUS";
-    private static boolean isBinaryLogStatementUndetermined = true;
-    private static String showBinaryLogStatement;
+    private String showBinaryLogStatement;
 
     /**
      * Creates a new connection using the supplied configuration.
@@ -100,36 +99,35 @@ public MySqlConnection(
     }
 
     public String probeShowBinaryLogStatement() {
-        if (isBinaryLogStatementUndetermined) {
+        if (showBinaryLogStatement != null) {
+            return showBinaryLogStatement;
+        }
+        try {
+            LOGGER.info("Probing binary log statement.");
             try {
-                LOGGER.info("Probing binary log statement.");
-                try {
-                    // Attempt to query
-                    query(MYSQL_NEW_SHOW_BINARY_LOG_STATEMENT, rs -> {});
-                    showBinaryLogStatement = MYSQL_NEW_SHOW_BINARY_LOG_STATEMENT;
-                } catch (SQLException e) {
-                    LOGGER.info(
-                            "Probing with {} failed, try {}. Caused by: {}",
-                            MYSQL_NEW_SHOW_BINARY_LOG_STATEMENT,
-                            MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT,
-                            e);
-                    query(MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT, rs -> {});
-                    showBinaryLogStatement = MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT;
-                }
+                // Attempt to query
+                query(MYSQL_NEW_SHOW_BINARY_LOG_STATEMENT, rs -> {});
+                showBinaryLogStatement = MYSQL_NEW_SHOW_BINARY_LOG_STATEMENT;
+            } catch (SQLException e) {
                 LOGGER.info(
-                        "Successfully found show binary log statement with `{}`.",
-                        showBinaryLogStatement);
-                // Make sure this check only happens once
-                isBinaryLogStatementUndetermined = false;
-            } catch (Exception e) {
-                showBinaryLogStatement = MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT;
-                LOGGER.warn(
-                        "Failed to probe binary log statement version. Fallback to `{}`.",
-                        showBinaryLogStatement,
+                        "Probing with {} failed, try {}. Caused by: {}",
+                        MYSQL_NEW_SHOW_BINARY_LOG_STATEMENT,
+                        MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT,
                         e);
+                query(MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT, rs -> {});
+                showBinaryLogStatement = MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT;
             }
+            LOGGER.info(
+                    "Successfully found show binary log statement with `{}`.",
+                    showBinaryLogStatement);
+            return showBinaryLogStatement;
+        } catch (Exception e) {
+            LOGGER.warn(
+                    "Failed to probe binary log statement version. Fallback to `{}`.",
+                    MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT,
+                    e);
+            return MYSQL_CLASSIC_SHOW_BINARY_LOG_STATEMENT;
         }
-        return showBinaryLogStatement;
     }
 
     /**