/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.connector.mysql;

import static org.junit.jupiter.api.Assertions.assertThrows;

import java.util.concurrent.atomic.AtomicReference;

import org.apache.kafka.connect.errors.ConnectException;
import org.junit.jupiter.api.Test;

import io.debezium.connector.binlog.BinlogReadOnlyIncrementalSnapshotIT;
import io.debezium.connector.binlog.junit.SkipWhenGtidModeIs;
import io.debezium.connector.mysql.jdbc.MySqlBinaryProtocolFieldReader;

@SkipWhenGtidModeIs(value = SkipWhenGtidModeIs.GtidMode.OFF, reason = "Read only connection requires GTID_MODE to be ON")
public class ReadOnlyIncrementalSnapshotIT extends BinlogReadOnlyIncrementalSnapshotIT<MySqlConnector> implements MySqlCommon {

    @Override
    protected Class<?> getFieldReader() {
        return MySqlBinaryProtocolFieldReader.class;
    }

    @Test
    @SkipWhenGtidModeIs(value = SkipWhenGtidModeIs.GtidMode.ON, reason = "Read only connection requires GTID_MODE to be ON")
    void shouldFailIfGtidModeIsOff() throws Exception {
        assertThrows(ConnectException.class, () -> {
            // Testing.Print.enable();
            populateTable();
            AtomicReference<Throwable> exception = new AtomicReference<>();
            startConnector((success, message, error) -> exception.set(error));
            waitForEngineShutdown();
            stopConnector();
            final Throwable e = exception.get();
            if (e != null) {
                throw (RuntimeException) e;
            }
        });
    }
}
