#!/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()