private SSLSocketFactory getBinlogSslSocketFactory(MySqlConnectorConfig connectorConfig,
            MySqlConnection connection) {
        String acceptedTlsVersion = connection.getSessionVariableForSslVersion();
        if (!isNullOrEmpty(acceptedTlsVersion)) {
            SSLMode sslMode = sslModeFor(connectorConfig.sslMode());

            // Keystore settings can be passed via system properties too so we need to read them
            final String password = System.getProperty("javax.net.ssl.keyStorePassword");
            final String keyFilename = System.getProperty("javax.net.ssl.keyStore");
            KeyManager[] keyManagers = null;
            if (keyFilename != null) {
                final char[] passwordArray = (password == null) ? null : password.toCharArray();
                try {
                    KeyStore ks = KeyStore.getInstance("JKS");
                    ks.load(new FileInputStream(keyFilename), passwordArray);

                    KeyManagerFactory kmf = KeyManagerFactory.getInstance("NewSunX509");
                    kmf.init(ks, passwordArray);

                    keyManagers = kmf.getKeyManagers();
                } catch (KeyStoreException | IOException | CertificateException | NoSuchAlgorithmException
                        | UnrecoverableKeyException e) {
                    throw new DebeziumException("Could not load keystore", e);
                }
            }

            // DBZ-1208 Resembles the logic from the upstream BinaryLogClient, only that
            // the accepted TLS version is passed to the constructed factory
            if (sslMode == SSLMode.PREFERRED || sslMode == SSLMode.REQUIRED) {
                final KeyManager[] finalKMS = keyManagers;
                return new DefaultSSLSocketFactory(acceptedTlsVersion) {

                    @Override
                    protected void initSSLContext(SSLContext sc)
                            throws GeneralSecurityException {
                        sc.init(finalKMS, new TrustManager[]{
                                new X509TrustManager() {

                                    @Override
                                    public void checkClientTrusted(
                                            X509Certificate[] x509Certificates,
                                            String s)
                                            throws CertificateException {
                                    }

                                    @Override
                                    public void checkServerTrusted(
                                            X509Certificate[] x509Certificates,
                                            String s)
                                            throws CertificateException {
                                    }

                                    @Override
                                    public X509Certificate[] getAcceptedIssuers() {
                                        return new X509Certificate[0];
                                    }
                                }
                        }, null);
                    }
                };
            } else {
                return new DefaultSSLSocketFactory(acceptedTlsVersion);
            }
        }

        return null;
    }
Handle a {@link EventType#XA_PREPARE} event.

@param event the database change data event to be processed; may not be null
@throws InterruptedException if this thread is interrupted while blocking
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java