diff --git a/docs/examples/python/connect_example.py b/docs/examples/python/connect_example.py
index 744f2aa49e..5abc767e85 100644
--- a/docs/examples/python/connect_example.py
+++ b/docs/examples/python/connect_example.py
@@ -8,7 +8,7 @@ def create_connection():
             user="root",
             password="taosdata",
             host="localhost",
-            port=6041,
+            port=6030,
         )
     except Exception as err:
         print(err)
diff --git a/docs/examples/python/connect_rest_example.py b/docs/examples/python/connect_rest_example.py
index c2b8f38431..266edfafe5 100644
--- a/docs/examples/python/connect_rest_example.py
+++ b/docs/examples/python/connect_rest_example.py
@@ -8,6 +8,9 @@ def create_connection():
                                 user="root",
                                 password="taosdata",
                                 timeout=30)
+        
+        print("Connection established")
+        
     except Exception as err:
         print(err)
     finally:
diff --git a/docs/examples/python/connect_websocket_examples.py b/docs/examples/python/connect_websocket_examples.py
index 3d93634bcc..939486de28 100644
--- a/docs/examples/python/connect_websocket_examples.py
+++ b/docs/examples/python/connect_websocket_examples.py
@@ -12,23 +12,27 @@ def create_connection():
         )
     except Exception as err:
         print(err)
-   
+
     return conn
- # ANCHOR_END: connect
+
+# ANCHOR_END: connect
 
 def create_db_table(conn):
-# ANCHOR: create_db
+    # ANCHOR: create_db
     try:
         conn.execute("CREATE DATABASE IF NOT EXISTS power")
         conn.execute("USE power")
-        conn.execute("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
+        conn.execute(
+            "CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
         conn.execute("CREATE TABLE  IF NOT EXISTS `d0` USING `meters` (groupId, location) TAGS(0, 'Los Angles')")
     except Exception as err:
         print(f'Exception {err}')
+
+
 # ANCHOR_END: create_db
 
 def insert(conn):
-# ANCHOR: insert
+    # ANCHOR: insert
     sql = """
     INSERT INTO 
     power.d1001 USING power.meters TAGS('California.SanFrancisco', 2)
@@ -39,22 +43,26 @@ def insert(conn):
     """
     try:
         inserted = conn.execute(sql)
-        assert inserted == 8
+        assert inserted == 4
     except Exception as err:
-        print(f'Exception111 {err}')
+        print(f'Exception {err}')
+
+
 # ANCHOR_END: insert
 
 def query(conn):
-# ANCHOR: query
+    # ANCHOR: query
     try:
         result = conn.query("select * from meters")
         num_of_fields = result.field_count
-        print(num_of_fields)
+        print(f"query field conunt is {num_of_fields}")
 
         for row in result:
             print(row)
     except Exception as err:
-        print(f'Exception {err}')
+        print(f'query Exception {err}')
+
+
 # ANCHOR_END: query
 
 if __name__ == "__main__":
diff --git a/docs/examples/python/create_db_ws.py b/docs/examples/python/create_db_ws.py
index 8c43af8db2..f60411f8e0 100644
--- a/docs/examples/python/create_db_ws.py
+++ b/docs/examples/python/create_db_ws.py
@@ -1,34 +1,76 @@
 import taosws
 
-conn = None
+db = "power"
 
-try:
-    conn = taosws.connect(user="root",
-                          password="taosdata",
-                          host="localhost",
-                          port=6041)
+def prepare():
+    conn = None
 
-    db = "power"
-    # create database
-    rowsAffected = conn.execute(f"CREATE DATABASE IF NOT EXISTS {db}")
-    assert rowsAffected == 0
+    try:
+        conn = taosws.connect(user="root",
+                              password="taosdata",
+                              host="192.168.1.98",
+                              port=6041)
 
-    # change database. 
-    rowsAffected = conn.execute(f"USE {db}")
-    assert rowsAffected == 0
-    
-    # create super table
-    rowsAffected = conn.execute(
-        "CREATE TABLE IF NOT EXISTS `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))"
-    )
-    assert rowsAffected == 0
 
-    # create table
-    rowsAffected = conn.execute("CREATE TABLE IF NOT EXISTS `d0` USING `meters` (groupid, location) TAGS(0, 'Los Angles')")
-    assert rowsAffected == 0
+        # create database
+        conn.execute(f"drop database if exists {db}")
+        conn.execute(f"create database {db}")
+    except Exception as err:
+        print(err)
+    finally:
+        if conn:
+            conn.close()
 
-except Exception as err:
-    print(err)
-finally:
-    if conn:
-        conn.close()
+def schemaless_insert():
+    conn = None
+
+    lineDemo = [
+        "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639"
+    ]
+
+    telnetDemo = ["metric_telnet 1707095283260 4 host=host0 interface=eth0"]
+
+    jsonDemo = [
+        '{"metric": "metric_json","timestamp": 1626846400,"value": 10.3, "tags": {"groupid": 2, "location": "California.SanFrancisco", "id": "d1001"}}'
+    ]
+
+    try:
+
+        conn = taosws.connect(user="root",
+                              password="taosdata",
+                              host="192.168.1.98",
+                              port=6041,
+                              database=db)
+
+        conn.schemaless_insert(
+            lines=lineDemo,
+            protocol=taosws.PySchemalessProtocol.Line,
+            precision=taosws.PySchemalessPrecision.Millisecond,
+            ttl=1,
+            req_id=1,
+        )
+
+        conn.schemaless_insert(
+            lines=telnetDemo,
+            protocol=taosws.PySchemalessProtocol.Telnet,
+            precision=taosws.PySchemalessPrecision.Microsecond,
+            ttl=1,
+            req_id=2,
+        )
+
+        conn.schemaless_insert(
+            lines=jsonDemo,
+            protocol=taosws.PySchemalessProtocol.Json,
+            precision=taosws.PySchemalessPrecision.Millisecond,
+            ttl=1,
+            req_id=3,
+        )
+    except Exception as err:
+        print(err)
+    finally:
+        if conn:
+            conn.close()
+
+if __name__ == '__main__':
+    prepare()
+    schemaless_insert()
\ No newline at end of file
diff --git a/docs/examples/python/insert_native.py b/docs/examples/python/insert_native.py
index ca59d8f395..ea8901f5b7 100644
--- a/docs/examples/python/insert_native.py
+++ b/docs/examples/python/insert_native.py
@@ -6,7 +6,7 @@ try:
     conn = taos.connect(user="root",
                           password="taosdata",
                           host="localhost",
-                          port=6041)
+                          port=6030)
 
     sql = """
         INSERT INTO 
@@ -17,7 +17,7 @@ try:
             VALUES (NOW + 1a, 10.30000, 218, 0.25000)
         """
     affectedRows = conn.execute(sql)
-    print("inserted into {affectedRows} rows to power.meters successfully.")
+    print(f"inserted into {affectedRows} rows to power.meters successfully.")
 
 except Exception as err:
     print(err)
diff --git a/docs/examples/python/insert_ws.py b/docs/examples/python/insert_ws.py
index 1d7aa7fc5d..d838e79670 100644
--- a/docs/examples/python/insert_ws.py
+++ b/docs/examples/python/insert_ws.py
@@ -16,8 +16,8 @@ try:
         power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
             VALUES (NOW + 1a, 10.30000, 218, 0.25000)
         """
-    inserted = conn.execute(sql)
-    print("inserted into {affectedRows} rows to power.meters successfully.")
+    affectedRows = conn.execute(sql)
+    print(f"inserted into {affectedRows} rows to power.meters successfully.")
 
 except Exception as err:
     print(err)
diff --git a/docs/examples/python/reqid_native.py b/docs/examples/python/reqid_native.py
index f6c27bceb3..715b1b39bc 100644
--- a/docs/examples/python/reqid_native.py
+++ b/docs/examples/python/reqid_native.py
@@ -15,8 +15,8 @@ try:
         power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
             VALUES (NOW + 1a, 10.30000, 218, 0.25000)
         """
-    inserted = conn.execute(sql, 1)
-    print("inserted into {affectedRows} rows to power.meters successfully.")
+    affectedRows = conn.execute(sql, 1)
+    print(f"inserted into {affectedRows} rows to power.meters successfully.")
 
     result = conn.query("SELECT ts, current, location FROM power.meters limit 100", 2)
     print(result)
diff --git a/docs/examples/python/reqid_ws.py b/docs/examples/python/reqid_ws.py
index 47f4737a81..1226ae8ad6 100644
--- a/docs/examples/python/reqid_ws.py
+++ b/docs/examples/python/reqid_ws.py
@@ -19,7 +19,7 @@ try:
             VALUES (NOW + 1a, 10.30000, 218, 0.25000)
         """
     affectedRows = conn.execute_with_req_id(sql, req_id=1)
-    print("inserted into {affectedRows} rows to power.meters successfully.")
+    print(f"inserted into {affectedRows} rows to power.meters successfully.")
 
     result = conn.query_with_req_id("SELECT ts, current, location FROM power.meters limit 100", req_id=2)
     num_of_fields = result.field_count
diff --git a/docs/examples/python/schemaless_ws.py b/docs/examples/python/schemaless_ws.py
index f6c1f1b0ac..845f1e65fd 100644
--- a/docs/examples/python/schemaless_ws.py
+++ b/docs/examples/python/schemaless_ws.py
@@ -1,46 +1,81 @@
 import taosws
 
-conn = None
+db = "power"
 
-lineDemo = [
-    "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639"
-]
+def prepare():
+    conn = None
 
-telnetDemo = ["metric_telnet 1707095283260 4 host=host0 interface=eth0"]
+    try:
+        conn = taosws.connect(user="root",
+                              password="taosdata",
+                              host="taosdata",
+                              port=6041)
 
-jsonDemo = [
-    '{"metric": "metric_json","timestamp": 1626846400,"value": 10.3, "tags": {"groupid": 2, "location": "California.SanFrancisco", "id": "d1001"}}'
-]
 
-try:
-    conn = taosws.connect(user="root",
-                          password="taosdata",
-                          host="localhost",
-                          port=6041)
+        # create database
+        conn.execute(f"drop database if exists {db}")
+        conn.execute(f"create database {db}")
+    except Exception as err:
+        print(err)
+        raise err
+    finally:
+        if conn:
+            conn.close()
 
-    conn.execute("CREATE DATABASE IF NOT EXISTS power")
-    conn = conn.execute("USE power")
+def schemaless_insert():
+    conn = None
 
-    conn.schemaless_insert(
-        lines=lineDemo,
-        protocol=taosws.PySchemalessProtocol.Line,
-        precision=taosws.PySchemalessPrecision.Millisecond
-    )
+    lineDemo = [
+        "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639"
+    ]
 
-    conn.schemaless_insert(
-        lines=telnetDemo,
-        protocol=taosws.PySchemalessProtocol.Telnet,
-        precision=taosws.PySchemalessPrecision.Microsecond
-    )
+    telnetDemo = ["metric_telnet 1707095283260 4 host=host0 interface=eth0"]
 
-    conn.schemaless_insert(
-        lines=jsonDemo,
-        protocol=taosws.PySchemalessProtocol.Json,
-        precision=taosws.PySchemalessPrecision.Millisecond
-    )
-except Exception as err:
-    print(err)
-finally:
-    if conn:
-        conn.close()
+    jsonDemo = [
+        '{"metric": "metric_json","timestamp": 1626846400,"value": 10.3, "tags": {"groupid": 2, "location": "California.SanFrancisco", "id": "d1001"}}'
+    ]
 
+    try:
+
+        conn = taosws.connect(user="root",
+                              password="taosdata",
+                              host="192.168.1.98",
+                              port=6041,
+                              database=db)
+
+        conn.schemaless_insert(
+            lines=lineDemo,
+            protocol=taosws.PySchemalessProtocol.Line,
+            precision=taosws.PySchemalessPrecision.Millisecond,
+            ttl=1,
+            req_id=1,
+        )
+
+        conn.schemaless_insert(
+            lines=telnetDemo,
+            protocol=taosws.PySchemalessProtocol.Telnet,
+            precision=taosws.PySchemalessPrecision.Microsecond,
+            ttl=1,
+            req_id=2,
+        )
+
+        conn.schemaless_insert(
+            lines=jsonDemo,
+            protocol=taosws.PySchemalessProtocol.Json,
+            precision=taosws.PySchemalessPrecision.Millisecond,
+            ttl=1,
+            req_id=3,
+        )
+    except Exception as err:
+        print(err)
+        raise err
+    finally:
+        if conn:
+            conn.close()
+
+if __name__ == '__main__':
+    try:
+        prepare()
+        schemaless_insert()
+    except Exception as err:
+        print(err)
\ No newline at end of file
diff --git a/docs/examples/python/stmt_native.py b/docs/examples/python/stmt_native.py
index c51167a299..f7ef1ba110 100644
--- a/docs/examples/python/stmt_native.py
+++ b/docs/examples/python/stmt_native.py
@@ -42,7 +42,7 @@ try:
         for j in range (numOfRow):
             timestamps.append(current + i)
             currents.append(random.random() * 30)
-            voltages.append(random.random(100, 300))
+            voltages.append(random.randint(100, 300))
             phases.append(random.random())
 
         params = taos.new_bind_params(4)
@@ -52,8 +52,7 @@ try:
         params[3].float(phases)
         stmt.bind_param_batch(params)
         stmt.execute()
-        affected = stmt.affected_rows()
-        print(f"table {tbname} insert {affected} rows.")
+        print(f"table {tbname} insert ok.")
 
 except Exception as err:
     print(err)
diff --git a/docs/examples/python/stmt_ws.py b/docs/examples/python/stmt_ws.py
index 360833208e..7f4ef75f74 100644
--- a/docs/examples/python/stmt_ws.py
+++ b/docs/examples/python/stmt_ws.py
@@ -38,10 +38,10 @@ try:
         currents = []
         voltages = []
         phases = []
-        for j in range (numOfRow):
+        for j in range(numOfRow):
             timestamps.append(current + i)
             currents.append(random.random() * 30)
-            voltages.append(random.random(100, 300))
+            voltages.append(random.randint(100, 300))
             phases.append(random.random())
 
         stmt.bind_param(
@@ -55,12 +55,12 @@ try:
 
         stmt.add_batch()
         rows = stmt.execute()
-        print(f"insert {rows} rows.")
-        
+        print(f"table {tbname} insert ok.")
+
 except Exception as err:
     print(err)
 finally:
     if stmt:
         stmt.close()
-    if conn:    
+    if conn:
         conn.close()
diff --git a/docs/examples/python/tmq_native.py b/docs/examples/python/tmq_native.py
index d43d8f4b8e..7770dda90d 100644
--- a/docs/examples/python/tmq_native.py
+++ b/docs/examples/python/tmq_native.py
@@ -1,8 +1,9 @@
 import taos
 
+
 def prepareMeta():
     conn = None
-    try:    
+    try:
         conn = taos.connect(
             host="localhost",
             user="root",
@@ -12,14 +13,14 @@ def prepareMeta():
 
         db = "power"
         topic = "topic_meters"
-        conn.execute(f"CREATE DATABASE IF EXISTS {db}")
+        conn.execute(f"CREATE DATABASE IF NOT EXISTS {db}")
 
         # change database. same as execute "USE db"
         conn.select_db(db)
 
         # create super table
         conn.execute(
-            "CREATE STABLE IF EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)"
+            "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)"
         )
 
         # ANCHOR: create_topic
@@ -37,15 +38,20 @@ def prepareMeta():
                 VALUES (NOW + 1a, 10.30000, 218, 0.25000)
             """
         affectedRows = conn.execute(sql)
-        print("inserted into {affectedRows} rows to power.meters successfully.")
+        print(f"inserted into {affectedRows} rows to power.meters successfully.")
     except Exception as err:
-        print("prepare meta err:{err}") 
-    finally
+        print(f"prepare meta err:{err}")
+        raise err
+    finally:
         if conn:
-            conn.close()      
+            conn.close()
+
+        # ANCHOR: create_consumer
+
 
-# ANCHOR: create_consumer
 from taos.tmq import Consumer
+
+
 def create_consumer():
     try:
         consumer = Consumer(
@@ -55,16 +61,18 @@ def create_consumer():
                 "td.connect.user": "root",
                 "td.connect.pass": "taosdata",
                 "enable.auto.commit": "true",
-                "auto.commit.interval.ms":"1000",
-                "auto.offset.reset": "latest", 
-                "td.connect.ip": "192.168.1.98",
-                "td.connect.port": "6041",
+                "auto.commit.interval.ms": "1000",
+                "auto.offset.reset": "latest",
+                "td.connect.ip": "localhost",
+                "td.connect.port": "6030",
             }
         )
+        return consumer
     except Exception as err:
-        print("Failed to poll data, err:{err}")
-        raise err                   
-# ANCHOR_END: create_consumer
+        print(f"Failed to poll data, err:{err}")
+        raise err
+    # ANCHOR_END: create_consumer
+
 
 # ANCHOR: subscribe
 def subscribe(consumer):
@@ -78,15 +86,17 @@ def subscribe(consumer):
                 if err is not None:
                     print(f"poll data error, {err}")
                     raise err
-                
-                val = res.value()
+
+                val = records.value()
                 if val:
                     for block in val:
-                        print(block.fetchall()) 
+                        print(block.fetchall())
 
     except Exception as err:
-        print("Failed to poll data, err:{err}")
+        print(f"Failed to poll data, err:{err}")
         raise err
+
+
 # ANCHOR_END: subscribe
 
 def commit_offset(consumer):
@@ -101,45 +111,49 @@ def commit_offset(consumer):
                 if err is not None:
                     print(f"poll data error, {err}")
                     raise err
-                
-                val = res.value()
+
+                val = records.value()
                 if val:
                     for block in val:
-                        print(block.fetchall()) 
+                        print(block.fetchall())
 
                 consumer.commit(records)
 
     except Exception as err:
-        print("Failed to poll data, err:{err}")
+        print(f"Failed to poll data, err:{err}")
         raise err
     # ANCHOR_END: commit_offset
 
+
 def seek_offset(consumer):
     # ANCHOR: assignment
     try:
         assignments = consumer.assignment()
-        for partition in assignments:
-            print("first data polled: {partition.offset}")
-            partition.offset = 0
-            consumer.seek(partition)
-            print("assignment seek to beginning successfully");
+        if assignments:
+            for partition in assignments:
+                print(f"first data polled: {partition.offset}")
+                partition.offset = 0
+                consumer.seek(partition)
+                print(f"assignment seek to beginning successfully");
     except Exception as err:
-        print("seek example failed; err:{err}")
+        print(f"seek example failed; err:{err}")
         raise err
     # ANCHOR_END: assignment
 
+
 # ANCHOR: unsubscribe
 def unsubscribe(consumer):
     try:
         consumer.unsubscribe()
     except Exception as err:
-        print("Failed to unsubscribe consumer. err:{err}")
-   
+        print(f"Failed to unsubscribe consumer. err:{err}")
+
+
 # ANCHOR_END: unsubscribe
 
 if __name__ == "__main__":
     consumer = None
-    try: 
+    try:
         prepareMeta()
         consumer = create_consumer()
         subscribe(consumer)
@@ -147,7 +161,7 @@ if __name__ == "__main__":
         commit_offset(consumer)
         unsubscribe(consumer)
     except Exception as err:
-        print("Failed to stmt consumer. err:{err}")
+        print(f"Failed to stmt consumer. err:{err}")
     finally:
         if consumer:
-            consumer.close()    
\ No newline at end of file
+            consumer.close()
\ No newline at end of file
diff --git a/docs/examples/python/tmq_websocket_example.py b/docs/examples/python/tmq_websocket_example.py
index 43ca725b56..400ea76234 100644
--- a/docs/examples/python/tmq_websocket_example.py
+++ b/docs/examples/python/tmq_websocket_example.py
@@ -1,24 +1,23 @@
 #!/usr/bin/python3
-from taosws
-
+import taosws
 def prepareMeta():
     conn = None
 
     try:
         conn = taosws.connect(user="root",
-                            password="taosdata",
-                            host="localhost",
-                            port=6041)
+                              password="taosdata",
+                              host="localhost",
+                              port=6041)
 
         db = "power"
         # create database
         rowsAffected = conn.execute(f"CREATE DATABASE IF NOT EXISTS {db}")
         assert rowsAffected == 0
 
-        # change database. 
+        # change database.
         rowsAffected = conn.execute(f"USE {db}")
         assert rowsAffected == 0
-        
+
         # create super table
         rowsAffected = conn.execute(
             "CREATE TABLE IF NOT EXISTS `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))"
@@ -26,7 +25,8 @@ def prepareMeta():
         assert rowsAffected == 0
 
         # create table
-        rowsAffected = conn.execute("CREATE TABLE IF NOT EXISTS `d0` USING `meters` (groupid, location) TAGS(0, 'Los Angles')")
+        rowsAffected = conn.execute(
+            "CREATE TABLE IF NOT EXISTS `d0` USING `meters` (groupid, location) TAGS(0, 'Los Angles')")
         assert rowsAffected == 0
 
         sql = """
@@ -37,15 +37,16 @@ def prepareMeta():
             power.d1002 USING power.meters (groupid, location) TAGS(3, 'California.SanFrancisco') 
                 VALUES (NOW + 1a, 10.30000, 218, 0.25000)
             """
-        inserted = conn.execute(sql)
-        print("inserted into {affectedRows} rows to power.meters successfully.")  
+        affectedRows = conn.execute(sql)
+        print(f"inserted into {affectedRows} rows to power.meters successfully.")
 
     except Exception as err:
-        print(err)
+        print(f"Failed to prepareMeta {err}")
+        raise err
     finally:
         if conn:
             conn.close()
-   
+
 
 # ANCHOR: create_consumer
 def create_consumer():
@@ -55,15 +56,17 @@ def create_consumer():
             "group.id": "group1",
             "client.id": "1",
             "auto.offset.reset": "latest",
-            "td.connect.ip": "192.168.1.98",
+            "td.connect.ip": "localhost",
             "td.connect.port": "6041",
             "enable.auto.commit": "true",
-            "auto.commit.interval.ms":"1000",
+            "auto.commit.interval.ms": "1000",
         })
         return consumer;
     except Exception as err:
         print(f"Failed to create websocket consumer, err:{err}");
         raise err
+
+
 # ANCHOR_END: create_consumer
 
 def seek_offset(consumer):
@@ -74,21 +77,23 @@ def seek_offset(consumer):
             topic = assignment.topic()
             print(f"topic: {topic}")
             for assign in assignment.assignments():
-                print(f"vg_id: {assign.vg_id()}, offset: {assign.offset()}, begin: {assign.begin()}, end: {assign.end()}")
+                print(
+                    f"vg_id: {assign.vg_id()}, offset: {assign.offset()}, begin: {assign.begin()}, end: {assign.end()}")
                 consumer.seek(topic, assign.vg_id(), assign.begin())
                 print("assignment seek to beginning successfully");
-    
+
     except Exception as err:
-        print("seek example failed; err:{err}")
+        print(f"seek example failed; err:{err}")
         raise err
     # ANCHOR_END: assignment
 
+
 # ANCHOR: subscribe
 def subscribe(consumer):
     try:
         consumer.subscribe(["topic_meters"])
         print("subscribe topics successfully")
-        for i in range(50):
+        for i in range(5):
             records = consumer.poll(timeout=1.0)
             if records:
                 for block in records:
@@ -96,16 +101,16 @@ def subscribe(consumer):
                         print(row)
 
     except Exception as err:
-        print("Failed to poll data, err:{err}")
+        print(f"Failed to poll data, err:{err}")
         raise err
+
+
 # ANCHOR_END: subscribe
 
 # ANCHOR: commit_offset
 def commit_offset(consumer):
     try:
-        consumer.subscribe(["topic_meters"])
-        print("subscribe topics successfully")
-        for i in range(50):
+        for i in range(5):
             records = consumer.poll(timeout=1.0)
             if records:
                 for block in records:
@@ -114,22 +119,25 @@ def commit_offset(consumer):
                 consumer.commit(records)
 
     except Exception as err:
-        print("Failed to poll data, err:{err}")
+        print(f"Failed to poll data, err:{err}")
         raise err
-# ANCHOR_END: commit_offset 
-#    
+
+
+# ANCHOR_END: commit_offset
+#
 # ANCHOR: unsubscribe
 def unsubscribe(consumer):
     try:
         consumer.unsubscribe()
     except Exception as err:
         print("Failed to unsubscribe consumer. err:{err}")
-   
+
+
 # ANCHOR_END: unsubscribe
 
 if __name__ == "__main__":
     consumer = None
-    try: 
+    try:
         prepareMeta()
         consumer = create_consumer()
         subscribe(consumer)
@@ -137,7 +145,7 @@ if __name__ == "__main__":
         commit_offset(consumer)
         unsubscribe(consumer)
     except Exception as err:
-        print("Failed to stmt consumer. err:{err}")
+        print(f"Failed to stmt consumer. err:{err}")
     finally:
         if consumer:
-            consumer.close() 
\ No newline at end of file
+            consumer.close()
\ No newline at end of file