#!/bin/env python
# -*- coding: utf-8 -*-
# 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.

############################################################################
#
#   @file test_sys_binlog.py
#   @date 2021/11/02 14:37:00
#   @brief This file is a test file for doris binlog load.
#
#############################################################################

"""
测试binlog load
MySQL需开启binlog功能
"""

import os
import sys
import time
import pymysql
sys.path.append("../")
from lib import palo_config
from lib import palo_client
from lib import palo_job
from lib import util
from data import binlog as DATA
client = None
config = palo_config.config
LOG = palo_client.LOG
L = palo_client.L

port = str(config.fe_query_port)
canal_ip = config.canal_ip
WAIT_TIME = 15

def setup_module():
    """
    setUp
    """
    global client
    global connect
    client = palo_client.get_client(config.fe_host, config.fe_query_port, user=config.fe_user,
                                    password=config.fe_password, http_port=config.fe_http_port)
    connect = pymysql.connect(host=config.mysql_host, port=config.mysql_port, user=config.canal_user, \
                                passwd=config.canal_password)
    # client.set_frontend_config('enable_create_sync_job', 'true')
    global destination
    destination = 'example_' + str(config.fe_query_port)


def mysql_execute(sql):
    """
    连接mysql执行导入语句
    """
    cursor = connect.cursor()
    try:
        LOG.info(L('mysql check sql', sql=sql))
        cursor.execute(sql)
        cursor.close()
        connect.commit()
    except Exception as error:
        assert False, "execute error. %s" % str(error)
        LOG.error(L("mysql execute error", error=str(error)))
    time.sleep(1)
    return cursor.fetchall()


def create_mysql_table(mysql_table_name, mysql_database_name, columns, key='PRIMARY KEY (k1)', new_database=True):
    """
    创建MySQL数据库和表
    """
    cursor = connect.cursor()
    if new_database:
        mysql_clean(mysql_database_name)
        sql = "CREATE DATABASE %s" % mysql_database_name
        try:
            cursor.execute(sql)
            LOG.info(L('CREATE MYSQL db succ', database=mysql_database_name))
        except Exception as error:
            LOG.error(L("CREATE MYSQL db error", host=config.mysql_host, database_name=mysql_database_name, \
                    error=error))
    connect.select_db(mysql_database_name)
    sql = "DROP TABLE IF EXISTS %s" % mysql_table_name
    cursor.execute(sql)
    sql = ''
    for column in columns:
        column_sql = '%s %s' % (column[0], column[1])
        if len(column) > 2:
            if column[2]:
                column_sql = '%s %s' % (column_sql, column[2])
        if len(column) > 3:
            column_sql = '%s DEFAULT "%s"' % (column_sql, column[3])
        sql = '%s %s,' % (sql, column_sql)
    sql = "CREATE table %s (%s %s)" % (mysql_table_name, sql, key)
    LOG.info(L('mysql table sql', sql=sql))
    try:
        cursor.execute(sql)
    except Exception as error:
        LOG.error(L("CREATE TABLE error", host=config.mysql_host, table_name=mysql_table_name, error=error))
        return False
    return True


def mysql_clean(mysql_database_name):
    """
    mysql drop database
    """
    sql = "DROP DATABASE IF EXISTS %s" % mysql_database_name
    cursor = connect.cursor()
    cursor.execute(sql)
    LOG.info(L("DROP MYSQL db", db=mysql_database_name))


def check(table_name, mysql_table_name):
    """验证doris与mysql数据同步"""
    sql_1 = "select * from %s order by k1" % table_name
    sql_2 = "select * from %s order by k1" % mysql_table_name
    cursor = connect.cursor()
    cursor.execute(sql_2)
    ret_2 = cursor.fetchall()
    time_limit = 0
    while time_limit < 60:
        ret_1 = client.execute(sql_1)
        if ret_1 == ret_2:
            return True
            break
        else:
            time_limit += 1
            time.sleep(2)
    return ret_1 == ret_2


def assertStop(ret, client, stop_job=None, info=''):
    """assert, if not stop binlog load"""
    if not ret and stop_job is not None:
        try:
            show_ret = client.show_sync_job()
            LOG.info(L('binlog load job info', ret=show_ret))
            client.stop_sync_job(stop_job)
        except Exception as e:
            print(str(e))
    assert ret, info


def check_data(data, table_name, job_name, fail_info, client=client):
    """验证binlog导入数据"""
    time_limit = 0
    try:
        while time_limit < 90:
            ret = client.verify(data, table_name)
            if ret:
                break
            else:
                time.sleep(2)
                time_limit += 1
    except Exception as e:
        LOG.info(L('get an error when check verify', msg=str(e)))
        client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, fail_info)


def test_binlog():
    """
    {
    "title": "test_sys_binlog:test_binlog",
    "describe": "创建binlog load任务，mysql执行insert, delete, update操作，验证doris中导入数据成功",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name 
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
                                 destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    #insert
    sql = open(DATA.binlog_sql_1, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_1, table_name, job_name, 'insert data failed', client)
    #delete
    sql = open(DATA.binlog_sql_2, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_2, table_name, job_name, 'delete data failed', client)
    #update
    sql = open(DATA.binlog_sql_3, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'update data failed', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_binlog_function():
    """
    {
    "title": "test_sys_binlog:test_binlog_function",
    "describe": "创建binlog load任务，mysql执行导入语句中包含函数，验证doris中导入数据成功",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_5, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_binlog_select():
    """
    {
    "title": "test_sys_binlog:test_binlog_select",
    "describe": "binlog load 过程中执行查询操作，验证查询结果正确",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = "select * from %s order by k1" % table_name
    ret = (client.execute(sql) == ())
    assertStop(ret, client, job_name, 'query data error')
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    ret = check(table_name, mysql_table_name)
    assertStop(ret, client, job_name, 'select data error')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_binlog_special():
    """
    {
    "title": "test_sys_binlog:test_binlog_special",
    "describe": "创建binlog load任务，mysql导入特殊值：NULL,随机数(支持decimal精度),时间，验证doris导入数据成功",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_2, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_2)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = 'insert into %s values (NULL, NULL, NULL, NULL, NULL)' % mysql_table_name
    util.assert_return(False, "Column 'k1' cannot be null", mysql_execute, sql)

    sql = 'insert into %s values (1, NULL, NULL, NULL, NULL)' % mysql_table_name
    mysql_execute(sql)
    ret = check(table_name, mysql_table_name)
    assertStop(ret, client, job_name, 'data load fail')

    sql = 'insert into %s values (2, FLOOR(RAND() * 10), "2021-08-18", "2021-08-18 00:05:10", 3.22), \
                (3, FLOOR(RAND() * 100), "2021-08-18", "2021-08-18 00:05:10", 3.22), \
                (4, FLOOR(RAND() * 1000), "2021-08-18", "2021-08-18 00:05:10", 3.22)' % mysql_table_name
    mysql_execute(sql)
    ret = check(table_name, mysql_table_name)
    assertStop(ret, client, job_name, 'data load fail')

    sql = 'insert into %s values (5, 20, "2021-08-18", "2021-08-18 00:05:10", RAND()), \
                (6, 20, "2021-08-18", "2021-08-18 00:05:10", RAND() * 10), \
                (7, 20, "2021-08-18", "2021-08-18 00:05:10", RAND() * 100)' % mysql_table_name
    mysql_execute(sql)
    ret = check(table_name, mysql_table_name)
    assertStop(ret, client, job_name, 'data load fail')

    sql = 'insert into %s values (8, 20, current_date, "2021-08-18 00:05:10", 3.22)' % mysql_table_name
    mysql_execute(sql)
    ret = check(table_name, mysql_table_name)
    assertStop(ret, client, job_name, 'data load fail')

    sql = 'insert into %s values (9, 20, "2021-08-18", now(), 3.22)' % mysql_table_name
    mysql_execute(sql)
    ret = check(table_name, mysql_table_name)
    assertStop(ret, client, job_name, 'data load fail')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_partition():
    """
    {
    "title": "test_sys_binlog:test_partition",
    "describe": "开启binlog load，doris增加、删除分区，mysql执行导入语句，验证数据可导入到已有分区",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['20', '40', '60', '80']
    partition_info = palo_client.PartitionInfo('k1', partition_name_list, partition_value_list)
    client.create_table(table_name, DATA.column_1, partition_info, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    #导入数据在doris中无分区，无法导入
    sql = open(DATA.binlog_sql_12, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    #add partition
    ret = client.add_partition(table_name, 'partition_e', '100')
    assertStop(ret, client, job_name, 'add partition failed')
    ret = client.verify(DATA.expected_file_3, table_name)
    assertStop(ret, client, job_name, 'data load fail')
    #drop partition without data
    ret = client.drop_partition(table_name, 'partition_d')
    assertStop(ret, client, job_name, 'Drop partition failed')
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    #drop partition with data
    ret = client.drop_partition(table_name, 'partition_a')
    assertStop(ret, client, job_name, 'drop partition failed')
    ret = not client.verify(DATA.expected_file_3, table_name)
    assertStop(ret, client, job_name, 'data load fail')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_mysql_transaction():
    """
    {
    "title": "test_sys_binlog:test_mysql_transaction",
    "describe": "开启binlog load，mysql提交事务，验证doris中数据导入成功",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql' + table_name
    transaction_table = mysql_table_name + '_t'
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    create_mysql_table(transaction_table, mysql_database_name, DATA.column_1, new_database=False)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql_1 = open(DATA.binlog_sql_1, 'r').read().format(mysql_table_name)
    sql_2 = open(DATA.binlog_sql_1, 'r').read().format(transaction_table)
    sql_3 = open(DATA.binlog_sql_2, 'r').read().format(mysql_table_name)
    sql_4 = open(DATA.binlog_sql_3, 'r').read().format(mysql_table_name)
    sql_5 = open(DATA.binlog_sql_2, 'r').read().format(transaction_table)
    sql = "begin; %s; %s; %s; %s; %s; commit;" % (sql_1, sql_2, sql_3, sql_4, sql_5)
    try:
        mysql_execute(sql)
        ret = True
    except:
        ret = False
    assertStop(ret, client, job_name, 'mysql transaction failed')
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_diff_schema_1():
    """
    {
    "title": "test_sys_binlog:test_diff_schema_1",
    "describe": "mysql与doris表结构不同，doris表有mysql表不存在的列，任务进行，但数据无法导入到doris",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_3, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_1, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    sql = 'select * from %s' % table_name
    ret = (client.execute(sql) == ())
    assertStop(ret, client, job_name, 'query data error')
    ret = client.get_sync_job_state(job_name) == 'RUNNING'
    assertStop(ret, client, job_name, 'sync job is not running')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_diff_schema_2():
    """
    {
    "title": "test_sys_binlog:test_diff_schema_2",
    "describe": "mysql与doris表结构不同，mysql表有doris表不存在的列，任务进行，但数据无法导入到doris",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_4, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_1, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    sql = 'select * from %s' % table_name
    ret = (client.execute(sql) == ())
    assertStop(ret, client, job_name, 'query data error')
    ret = client.get_sync_job_state(job_name) == 'RUNNING'
    assertStop(ret, client, job_name, 'sync job is not running')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_alter_mysql_schema():
    """
    {
    "title": "test_sys_binlog:test_alter_mysql_schema",
    "describe": "创建binlog load任务，修改mysql表结构，mysql执行导入语句，验证数据按doris列导入",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name + str(int(time.time()))
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #add column
    sql = "alter table %s add column k8 int(10)" % mysql_table_name
    mysql_execute(sql)
    sql = open(DATA.binlog_sql_9, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load fail')
    #drop new column
    sql = "alter table %s drop column k8" % mysql_table_name
    mysql_execute(sql)
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data([DATA.expected_file_5, DATA.expected_file_7], table_name, job_name, 'data load fail', client)
    #drop column
    sql = "alter table %s drop column k7" % mysql_table_name
    mysql_execute(sql)
    sql = open(DATA.binlog_sql_10, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify([DATA.expected_file_5, DATA.expected_file_7], table_name)
    assertStop(ret, client, job_name, 'data load fail')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_alter_doris_schema():
    """
    {
    "title": "test_sys_binlog:test_alter_doris_schema",
    "describe": "创建binlog load任务，修改doris表结构，mysql执行导入语句，验证数据按doris列导入",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #add column
    client.pause_sync_job(job_name)
    client.schema_change(table_name, add_column_list=["k8 int"], is_wait=True)
    client.resume_sync_job(job_name)
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_8, table_name, job_name, 'data load fail', client)
    #drop new column
    client.schema_change(table_name, drop_column_list=["k8"], is_wait=True)
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data([DATA.expected_file_5, DATA.expected_file_6, DATA.expected_file_7], table_name, job_name, \
                'data load fail', client)
    #drop column
    client.schema_change(table_name, drop_column_list=["k7"], is_wait=True)
    sql = open(DATA.binlog_sql_11, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_9, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_mysql_primary_key():
    """
    {
    "title": "test_sys_binlog:test_mysql_promary_key",
    "describe": "创建binlog load任务，mysql操作primary索引表导入数据，验证支持binlog导入数据到doris",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_mysql_unique_key():
    """
    {
    "title": "test_sys_binlog:test_mysql_unique_key",
    "describe": "创建binlog load任务，mysql操作unique索引表导入数据，验证支持binlog导入数据到doris",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1, 'UNIQUE (k1)')
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_mysql_index_key():
    """
    {
    "title": "test_sys_binlog:test_mysql_index_key",
    "describe": "创建binlog load任务，mysql操作index索引表导入数据，验证支持binlog导入数据到doris",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1, 'INDEX (k1)')
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_pause_resume():
    """
    {
    "title": "test_sys_binlog:test_pause_resume",
    "describe": "创建binlog load任务，验证暂停任务后重启任务成功",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret =  (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #pause sync job
    ret = client.pause_sync_job(job_name)
    assertStop(ret, client, job_name, 'pause sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #resume sync job
    ret = client.resume_sync_job(job_name)
    assertStop(ret, client, job_name, 'resume sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    expected_file_list = [DATA.expected_file_5, DATA.expected_file_6, DATA.expected_file_7]
    check_data(expected_file_list, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_create_same_job():
    """
    {
    "title": "test_sys_binlog:test_create_same_job",
    "describe": "验证创建相同的binlog load任务失败",
    "tag": "p0,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    util.assert_return(False, 'already exists', client.create_sync_job, table_name, database_name, mysql_table_name, \
            mysql_database_name, job_name, canal_ip, destination=destination)
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_resume_after_create():
    """
    {
    "title": "test_sys_binlog:test_resume_after_create",
    "describe": "创建binlog load任务后，重启任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    util.assert_return(False, 'There is no paused job', client.resume_sync_job, job_name)
    ret = client.get_sync_job_state(job_name) == 'RUNNING'
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_pause_same_job():
    """
    {
    "title": "test_sys_binlog:test_pause_same_job",
    "describe": "创建binlog load任务，验证暂停同一个任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    ret = client.pause_sync_job(job_name)
    assertStop(ret, client, job_name, 'pause sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    util.assert_return(False, "There is no running job", client.pause_sync_job, job_name)
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    sql = "select * from %s" % table_name
    ret = (client.execute(sql) == ())
    assertStop(ret, client, job_name, 'query data error')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_create_after_pause():
    """
    {
    "title": "test_sys_binlog:test_create_after_pause",
    "describe": "创建binlog load任务，暂停任务后再次创建该任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    ret = client.pause_sync_job(job_name)
    assertStop(ret, client, job_name, 'pause sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    util.assert_return(False, "already exists", client.create_sync_job, table_name, database_name, mysql_table_name, \
                    mysql_database_name, job_name, canal_ip)
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    sql = "select * from %s" % table_name
    ret = (client.execute(sql) == ())
    assertStop(ret, client, job_name, 'query data error')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_resume_same_job():
    """
    {
    "title": "test_sys_binlog:test_resume_same_job",
    "describe": "创建binlog load任务，暂停任务，重启任务，验证再次重启失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    ret = client.pause_sync_job(job_name)
    assertStop(ret, client, job_name, 'pause sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    ret = client.resume_sync_job(job_name)
    assertStop(ret, client, job_name, 'resume sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    util.assert_return(False, "There is no paused job", client.resume_sync_job, job_name)
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_create_after_resume():
    """
    {
    "title": "test_sys_binlog:test_create_after_resume",
    "describe": "创建binlog load任务，暂停任务，重启任务，验证再次创建该任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    ret = client.pause_sync_job(job_name)
    assertStop(ret, client, job_name, 'pause sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    ret = client.resume_sync_job(job_name)
    assertStop(ret, client, job_name, 'resume sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    util.assert_return(False, "already exists", client.create_sync_job, table_name, database_name, mysql_table_name, \
            mysql_database_name, job_name, canal_ip, destination=destination)
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_pause_and_resume():
    """
    {
    "title": "test_sys_binlog:test_pause_and_resume",
    "describe": "创建binlog load任务，验证多次暂停、重启任务成功",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    count = 10
    while count > 0:
        ret = client.pause_sync_job(job_name)
        assertStop(ret, client, job_name, 'pause sync job failed')
        ret = (client.get_sync_job_state(job_name) == 'PAUSED')
        assertStop(ret, client, job_name, 'sync job state error')
        ret = client.resume_sync_job(job_name)
        assertStop(ret, client, job_name, 'resume sync job failed')
        ret = (client.get_sync_job_state(job_name) == 'RUNNING')
        assertStop(ret, client, job_name, 'sync job state error')
        time.sleep(1)
        count -= 1
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_sync_job_info():
    """
    {
    "title": "test_sys_binlog:test_sync_job_info",
    "describe": "创建binlog load任务，验证show sync job info",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    canal_port = '11111'
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    sync_job_list = client.show_sync_job()
    sync_channel = '%s.%s->%s' % (mysql_database_name, mysql_table_name, table_name)
    sync_job_config = 'address:%s:%s,destination:%s,batchSize:8192' % (canal_ip, canal_port, destination)
    for sync_job in sync_job_list:
        sync_job_info = palo_job.SyncJobInfo(sync_job)
        if sync_job_info.get_job_name() == job_name:
            ret = (sync_job_info.get_state() == 'RUNNING')
            assertStop(ret, client, job_name, 'sync job state error')
            ret = (sync_job_info.get_channel() == sync_channel)
            assertStop(ret, client, job_name, 'sync channel error')
            ret = (sync_job_info.get_job_config() == sync_job_config)
            assertStop(ret, client, job_name, 'sync job config error')
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_tables_in_one_job():
    """
    {
    "title": "test_sys_binlog:test_tables_in_one_job",
    "describe": "创建单个binlog load任务，任务中包括多对表的对应关系，mysql执行导入语句，验证数据导入doris成功",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    mysql_database_name = 'm_' + database_name + port
    table_names = []
    mysql_table_names = []
    mysql_database_names = [mysql_database_name for i in range(3)]
    for i in ['_1', '_2', '_3']:
        table_name_s = table_name + i
        client.create_table(table_name_s, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
        table_names.append(table_name_s)
        mysql_table_name = 'mysql_' + table_name_s
        if i == '_1':
            create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
        else:
            create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1, new_database=False)
        mysql_table_names.append(mysql_table_name)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_names, database_name, mysql_table_names, mysql_database_names, job_name, \
            canal_ip, destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_names[0])
    mysql_execute(sql)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_names[1])
    mysql_execute(sql)
    sql = open(DATA.binlog_sql_1, 'r').read().format(mysql_table_names[2])
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    check_data(DATA.expected_file_3, table_names[0], job_name, 'data load fail', client)
    check_data(DATA.expected_file_3, table_names[1], job_name, 'data load fail', client)
    check_data(DATA.expected_file_1, table_names[2], job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_tables_in_jobs():
    """
    {
    "title": "test_sys_binlog:test_tables_in_jobs",
    "describe": "创建多个binlog load任务，任务中包含多对表的对应关系，mysql执行导入语句，验证数据导入doris成功",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    mysql_database_name = 'm_' + database_name + port
    destination_list = [destination + '_1', destination + '_2', destination + '_3']
    count = 0
    job_name_list = []
    mysql_table_name_list = []
    table_name_list = []
    for i in ['_1', '_2', '_3']:
        table_name_s = table_name + i
        table_name_list.append(table_name_s)
        client.create_table(table_name_s, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
        mysql_table_name = 'mysql_' + table_name_s
        if i == '_1':
            create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
        else:
            create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1, new_database=False)
        mysql_table_name_list.append(mysql_table_name)
        job_name = 'job_' + table_name_s
        job_name_list.append(job_name)
        ret = client.create_sync_job(table_name_s, database_name, mysql_table_name, mysql_database_name, job_name, \
                canal_ip, destination=destination_list[count])
        assertStop(ret, client, job_name, 'create sync job failed')
        ret = (client.get_sync_job_state(job_name) == 'RUNNING')
        assertStop(ret, client, job_name, 'sync job state error')
        count += 1
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name_list[0])
    mysql_execute(sql)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name_list[1])
    mysql_execute(sql)
    sql = open(DATA.binlog_sql_1, 'r').read().format(mysql_table_name_list[2])
    mysql_execute(sql)
    for i in range(3):
        ret = (client.get_sync_job_state(job_name_list[i]) == 'RUNNING')
        assertStop(ret, client, job_name_list[i], 'sync job state error')
    check_data(DATA.expected_file_3, table_name_list[0], job_name_list[0], 'data load fail', client)
    check_data(DATA.expected_file_3, table_name_list[1], job_name_list[1], 'data load fail', client)
    check_data(DATA.expected_file_1, table_name_list[2], job_name_list[2], 'data load fail', client)
    for i in range(3):
        client.stop_sync_job(job_name_list[i])
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_stop_after_create():
    """
    {
    "title": "test_sys_binlog:test_stop_after_create",
    "describe": "创建binlog load任务，验证终止任务成功",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_stop_after_pause():
    """
    {
    "title": "test_sys_binlog:test_stop_after_pause",
    "describe": "创建binlog load任务，验证暂停任务后终止任务成功",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #pause sync job
    ret = client.pause_sync_job(job_name)
    assertStop(ret, client, job_name, 'pause sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_stop_after_resume():
    """
    {
    "title": "test_sys_binlog:test_stop_after_resume",
    "describe": "创建binlog load任务，暂停任务，重启任务，验证终止任务成功",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #pause sync job
    ret = client.pause_sync_job(job_name)
    assertStop(ret, client, job_name, 'pause sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'PAUSED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    #resume sync job
    ret = client.resume_sync_job(job_name)
    assertStop(ret, client, job_name, 'resume sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    expected_file_list = [DATA.expected_file_5, DATA.expected_file_6, DATA.expected_file_7]
    check_data(expected_file_list, table_name, job_name, 'data load fail', client)
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_11, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(expected_file_list, table_name)
    assertStop(ret, client, job_name, 'data load error')
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_stop_after_stop():
    """
    {
    "title": "test_sys_binlog:test_stop_after_stop",
    "describe": "创建binlog load任务，终止任务，验证再次终止任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    #stop sync job again
    util.assert_return(False, 'There is no uncompleted job', client.stop_sync_job, job_name)
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_create_same_job_after_stop():
    """
    {
    "title": "test_sys_binlog:test_create_same_job_after_stop",
    "describe": "创建binlog load任务，终止任务，验证创建不同名但表对应与之前相同的任务成功，数据能够继续导入",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    #create same sync
    job_name_2 = job_name + '_2'
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name_2, \
            canal_ip, destination=destination)
    assertStop(ret, client, job_name_2, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name_2) == 'RUNNING')
    assertStop(ret, client, job_name_2, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    expected_file_list = [DATA.expected_file_5, DATA.expected_file_6, DATA.expected_file_7]
    check_data(expected_file_list, table_name, job_name_2, 'data load fail', client)
    client.stop_sync_job(job_name_2)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_create_same_name_job_after_stop():
    """
    {
    "title": "test_sys_binlog:test_create_same_name_job_after_stop",
    "describe": "创建binlog load任务，终止任务，验证创建同名任务成功，执行mysql导入语句，验证doris数据导入正确",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    #create same sync job
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, \
            canal_ip, destination=destination, is_wait=False)
    assertStop(ret, client, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    expected_file_list = [DATA.expected_file_5, DATA.expected_file_6, DATA.expected_file_7]
    check_data(expected_file_list, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_pause_after_stop():
    """
    {
    "title": "test_sys_binlog:test_pause_after_stop",
    "describe": "创建binlog load任务，终止任务，验证暂停任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    #pause sync job
    util.assert_return(False, 'There is no running job', client.pause_sync_job, job_name)
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_resume_after_stop():
    """
    {
    "title": "test_sys_binlog:test_resume_after_stop",
    "describe": "创建binlog load任务，终止任务，验证重启任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_6, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_5, table_name, job_name, 'data load fail', client)
    #stop sync job
    ret = client.stop_sync_job(job_name)
    assertStop(ret, client, job_name, 'stop sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_7, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    #resume sync job
    util.assert_return(False, 'There is no paused job', client.resume_sync_job, job_name)
    ret = (client.get_sync_job_state(job_name) == 'CANCELLED')
    assertStop(ret, client, job_name, 'sync job state error')
    sql = open(DATA.binlog_sql_8, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    time.sleep(WAIT_TIME)
    ret = client.verify(DATA.expected_file_5, table_name)
    assertStop(ret, client, job_name, 'data load error')
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_column():
    """
    {
    "title": "test_sys_binlog:test_column",
    "describe": "创建binlog load任务，指定列映射，mysql执行导入命令，验证doris数据导入",
    "tag": "p0,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_5, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    columns = ['k1', 'k6', 'k3', 'k4', 'k2', 'k7', 'k5']
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            columns=columns, destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_10, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_add_nullable_column():
    """
    {
    "title": "test_sys_binlog:test_add_nullable_column",
    "describe": "创建binlog load任务，指定列映射，doris增加nullable列，mysql执行导入命令，验证数据导入到doris",
    "tag": "p1,system"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_5, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    columns = ['k1', 'k6', 'k3', 'k4', 'k2', 'k7', 'k5']
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            columns=columns, destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    time.sleep(WAIT_TIME)
    client.schema_change(table_name, add_column_list=["k8 int after k1"], is_wait=True)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_11, table_name, job_name, 'data load fail', client)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_same_destination():
    """
    {
    "title": "test_sys_binlog:test_same_destination",
    "describe": "canal中每一个的destination仅对应一个binlog load任务，重复使用destination，验证创建binlog load任务失败",
    "tag": "p1,system,fuzz"
    }
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)
    client.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, canal_ip, \
            destination=destination)
    assertStop(ret, client, job_name, 'create sync job failed')
    ret = (client.get_sync_job_state(job_name) == 'RUNNING')
    assertStop(ret, client, job_name, 'sync job state error')
    #使用重复的destination创建binlog load任务失败
    table_name_s = table_name + '_s'
    mysql_table_name_s = mysql_table_name + '_s'
    client.create_table(table_name_s, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    create_mysql_table(mysql_table_name_s, mysql_database_name, DATA.column_1, new_database=False)
    job_name_s = job_name + '_s'
    util.assert_return(False, 'conflict destination', client.create_sync_job, table_name_s, database_name, \
                    mysql_table_name_s, mysql_database_name, job_name_s, canal_ip, destination=destination)
    client.stop_sync_job(job_name)
    client.clean(database_name)
    mysql_clean(mysql_database_name)


def test_observer_fe():
    """
    {
    "title": "test_sys_binlog:test_observer_fe",
    "describe": "连接observer fe，创建binlog load 任务，mysql执行导入语句，验证doris导入数据",
    "tag": "p1,system"
    }
    """
    client_observer = palo_client.get_client(config.fe_observer_list[0], config.fe_query_port, user=config.fe_user, \
                    password=config.fe_password, http_port=config.fe_http_port)
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client_observer.clean(database_name)
    client_observer.create_database(database_name)
    client_observer.use(database_name)
    client_observer.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client_observer.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, \
            canal_ip, destination=destination)
    assertStop(ret, client_observer, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client_observer)
    client_observer.stop_sync_job(job_name)
    client_observer.clean(database_name)
    mysql_clean(mysql_database_name)


def test_follower_fe():
    """
    {
    "title": "test_sys_binlog:test_follower_fe",
    "describe": "连接follower fe，创建binlog load 任务，mysql执行导入语句，验证doris导入数据",
    "tag": "p1,system"
    }
    """
    client_follower = palo_client.get_client(config.fe_follower_list[0], config.fe_query_port, user=config.fe_user, \
                    password=config.fe_password, http_port=config.fe_http_port)
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name, table_name=table_name, index_name=index_name))
    client_follower.clean(database_name)
    client_follower.create_database(database_name)
    client_follower.use(database_name)
    client_follower.create_table(table_name, DATA.column_1, set_null=True, keys_desc='UNIQUE KEY(k1)')
    mysql_database_name = 'm_' + database_name + port
    mysql_table_name = 'mysql_' + table_name
    create_mysql_table(mysql_table_name, mysql_database_name, DATA.column_1)
    job_name = 'job_' + table_name
    ret = client_follower.create_sync_job(table_name, database_name, mysql_table_name, mysql_database_name, job_name, \
            canal_ip, destination=destination)
    assertStop(ret, client_follower, job_name, 'create sync job failed')
    time.sleep(WAIT_TIME)
    sql = open(DATA.binlog_sql_4, 'r').read().format(mysql_table_name)
    mysql_execute(sql)
    check_data(DATA.expected_file_3, table_name, job_name, 'data load fail', client_follower)
    client_follower.stop_sync_job(job_name)
    client_follower.clean(database_name)
    mysql_clean(mysql_database_name)


def teardown_module():
    """
    tearDown
    """
    connect.close()