Merge branch 'test/ts-3497' of https://github.com/taosdata/TDengine into feature/TS-3495

tests/system-test/7-tmq/tmqParamsTest.py

+
+import taos
+import sys
+import time
+import socket
+import os
+import threading
+import math
+from taos.tmq import Consumer
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+from util.common import *
+sys.path.append("./7-tmq")
+from tmqCommon import *
+
+class TDTestCase:
+    updatecfgDict = {'debugFlag': 135}
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug(f"start to excute {__file__}")
+        tdSql.init(conn.cursor())
+        self.wal_retention_period1 = 3600
+        self.wal_retention_period2 = 1
+        self.commit_value_list = ["true", "false"]
+        self.commit_value_list = ["true"]
+        self.offset_value_list = ["", "earliest", "latest", "none"]
+        self.tbname_value_list = ["true", "false"]
+        self.snapshot_value_list = ["true", "false"]
+
+        # self.commit_value_list = ["true"]
+        # self.offset_value_list = ["latest"]
+        # self.offset_value_list = ["earliest"]
+        # self.tbname_value_list = ["true"]
+
+    def tmqParamsTest(self):
+        tdLog.printNoPrefix("======== test case 1: ")
+        paraDict = {'dbName':     'db1',
+                    'dropFlag':   1,
+                    'vgroups':    4,
+                    'stbName':    'stb',
+                    'colSchema':   [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
+                    'tagSchema':   [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
+                    'ctbPrefix':  'ctb',
+                    'ctbNum':     1,
+                    'rowsPerTbl': 10000,
+                    'batchNum':   10,
+                    'startTs':    1640966400000,  # 2022-01-01 00:00:00.000
+                    'auto_commit_interval': "100"}
+
+        
+        start_group_id = 1
+        for snapshot_value in self.snapshot_value_list:
+            for commit_value in self.commit_value_list:
+                for offset_value in self.offset_value_list:
+                    for tbname_value in self.tbname_value_list:
+                        topic_name = 'topic1'
+                        tmqCom.initConsumerTable()
+                        tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=4,replica=1)
+                        tdLog.info("create stb")
+                        tdCom.create_stable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"], column_elm_list=paraDict['colSchema'], tag_elm_list=paraDict['tagSchema'])
+                        tdLog.info("create ctb")
+                        tdCom.create_ctable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"],tag_elm_list=paraDict['tagSchema'],count=paraDict["ctbNum"], default_ctbname_prefix=paraDict['ctbPrefix'])
+                        tdLog.info("insert data")
+                        tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"])
+                        
+                        
+                        tdLog.info("create topics from stb with filter")
+                        queryString = "select ts, log(c1), ceil(pow(c1,3)) from %s.%s where c1 %% 7 == 0" %(paraDict['dbName'], paraDict['stbName'])
+                        sqlString = "create topic %s as %s" %(topic_name, queryString)
+                        print("----", snapshot_value)
+                        tdSql.query(f'select * from information_schema.ins_databases')
+                        db_wal_retention_period_list = list(map(lambda x:x[-8] if x[0] == paraDict['dbName'] else None, tdSql.queryResult))
+                        print("---db_wal_retention_period_list", db_wal_retention_period_list)
+                        for i in range(len(db_wal_retention_period_list)):
+                            if db_wal_retention_period_list[0] is None or db_wal_retention_period_list[-1] is None:
+                                db_wal_retention_period_list.remove(None)
+                        if snapshot_value =="true":
+                            if db_wal_retention_period_list[0] != self.wal_retention_period2:
+                                tdSql.execute(f"alter database {paraDict['dbName']} wal_retention_period {self.wal_retention_period2}")
+                                time.sleep(self.wal_retention_period2+1)
+                                tdSql.execute(f'flush database {paraDict["dbName"]}')
+                        else:
+                            print("iinininini")
+                            if db_wal_retention_period_list[0] != self.wal_retention_period1:
+                                tdSql.execute(f"alter database {paraDict['dbName']} wal_retention_period {self.wal_retention_period1}")
+                        tdLog.info("create topic sql: %s"%sqlString)
+                        tdSql.execute(sqlString)
+                        tdSql.query(queryString)
+                        expected_res = tdSql.queryRows
+                        group_id = "csm_" + str(start_group_id)
+                        consumer_dict = {
+                            "group.id": group_id,
+                            "td.connect.user": "root",
+                            "td.connect.pass": "taosdata",
+                            "auto.commit.interval.ms": paraDict["auto_commit_interval"],
+                            "enable.auto.commit": commit_value,
+                            "auto.offset.reset": offset_value,
+                            "experimental.snapshot.enable": snapshot_value,
+                            "msg.with.table.name": tbname_value
+                        }
+                        print(consumer_dict)
+                        consumer_commit = 1 if consumer_dict["enable.auto.commit"] == "true" else 0
+                        consumer_tbname = 1 if consumer_dict["msg.with.table.name"] == "true" else 0
+                        # consumer_snapshot = 1 if consumer_dict["experimental.snapshot.enable"] == "true" else 0
+                        consumer_ret = "earliest" if offset_value == "" else offset_value
+                        expected_parameters=f'tbname:{consumer_tbname},commit:{consumer_commit},interval:{paraDict["auto_commit_interval"]},reset:{consumer_ret}'
+                        if len(offset_value) == 0:
+                            del consumer_dict["auto.offset.reset"]
+                        consumer = Consumer(consumer_dict)
+                        consumer.subscribe([topic_name])
+                        stop_flag = 0
+                        # try:
+                        #     while True:
+                        #         res = consumer.poll(1)
+                                
+                        #         tdSql.query('show consumers;')
+                        #         consumer_info = tdSql.queryResult[0][-1]
+                        #         if not res:
+                        #             break
+                        #         # err = res.error()
+                        #         # if err is not None:
+                        #         #     raise err
+                        #         # val = res.value()
+                        #         # for block in val:
+                        #         #     print(block.fetchall())
+                        try:
+                            while True:
+                                res = consumer.poll(1)
+                                
+                                tdSql.query('show consumers;')
+                                consumer_info = tdSql.queryResult[0][-1]
+                                if offset_value == "latest":
+                                    if not res and stop_flag == 1:
+                                        break
+                                else:
+                                    if not res:
+                                        break
+                                # err = res.error()
+                                # if err is not None:
+                                #     raise err
+                                # val = res.value()
+                                # for block in val:
+                                #     print(block.fetchall())
+                                if offset_value == "latest" and stop_flag == 0:
+                                    tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],int(round(time.time()*1000)))
+                                    stop_flag = 1
+                        finally:
+                            consumer.unsubscribe()
+                            consumer.close()
+                        tdSql.checkEqual(consumer_info, expected_parameters)
+                        start_group_id += 1
+                        tdSql.query('show subscriptions;')
+                        subscription_info = tdSql.queryResult
+                        print(subscription_info)
+                        if snapshot_value == "true":
+                            if offset_value != "earliest":
+                                pass
+                            else:
+                                if offset_value != "none":
+                                    offset_value_str = ",".join(list(map(lambda x: x[-2], subscription_info)))
+                                    print(offset_value_str)
+                                    tdSql.checkEqual("snapshot" in offset_value_str, True)
+                                    rows_value_list  = list(map(lambda x: int(x[-1]), subscription_info))
+                                    tdSql.checkEqual(sum(rows_value_list), expected_res)
+                                else:
+                                    offset_value_list = list(map(lambda x: x[-2], subscription_info))
+                                    tdSql.checkEqual(offset_value_list, [None]*len(subscription_info))
+                                    print(offset_value_list)
+                                    rows_value_list  = list(map(lambda x: x[-1], subscription_info))
+                                    tdSql.checkEqual(rows_value_list, [None]*len(subscription_info))
+                        else:
+                            print("====offset_value----", offset_value)
+                            if offset_value != "none":
+                                offset_value_list = list(map(lambda x: int(x[-2].replace("log:", "")), subscription_info))
+                                tdSql.checkEqual(sum(offset_value_list) > 0, True)
+                                print(offset_value_list)
+                                rows_value_list  = list(map(lambda x: int(x[-1]), subscription_info))
+                                tdSql.checkEqual(sum(rows_value_list), expected_res)
+                            else:
+                                offset_value_list = list(map(lambda x: x[-2], subscription_info))
+                                tdSql.checkEqual(offset_value_list, [None]*len(subscription_info))
+                                print(offset_value_list)
+                                rows_value_list  = list(map(lambda x: x[-1], subscription_info))
+                                tdSql.checkEqual(rows_value_list, [None]*len(subscription_info))
+                            # tdSql.checkEqual(sum(rows_value_list), expected_res)
+                            # if offset_value == "latest":
+                            #     tdSql.checkEqual(sum(rows_value_list), expected_res)
+                            # else:
+                            #     tdSql.checkEqual(sum(rows_value_list), expected_res)
+                        tdSql.execute(f"drop topic if exists {topic_name}")
+                        tdSql.execute(f'drop database if exists {paraDict["dbName"]}')
+
+    def run(self):
+        self.tmqParamsTest()
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+event = threading.Event()
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())