finetuning

kafka_setup/src/main.py

@@ -69,16 +69,18 @@ def send_data(producer: KafkaProducer):
             with_metadata = random.choice([0, 1])
             if with_metadata == 0:
                 msg.pop("metadata")
+            print(i, msg.get("metadata"), msg["reporter"], msg["reported"])
         elif version == "v2.0.0":
             v2 = {
                 "metadata": {"version": "v2.0.0"},
                 "reporter_id": random.choice(players).replace("player", ""),
                 "reported_id": random.choice(players).replace("player", ""),
             }
             msg.update(v2)
+            print(i, msg.get("metadata"), msg["reporter_id"], msg["reported_id"])
         assert "reporter" in msg or "reported_id" in msg
         producer.send(topic="report", value=msg)
-        print(i, msg["metadata"], msg["reporting_id"], msg["reported_id"])
+
     print("Data insertion completed.")
 
 

src/main.py

@@ -92,7 +92,7 @@ async def queue_to_batch(queue: Queue, max_len: int = None) -> list:
 
 
 async def insert_batch(valid_report_queue: Queue):
-    INSERT_INTERVAL_SEC = 20
+    INSERT_INTERVAL_SEC = 60
     last_time = time.time()
     batch = []
     while True:
@@ -221,7 +221,6 @@ async def process_data(report_queue: Queue, player_cache: SimpleALRUCache):
 
 async def main():
     report_queue = Queue(maxsize=500)
-    valid_report_queue = Queue()
     await producer.start_engine(topic="report")
     await consumer.start_engine(topics=["report"])
 
@@ -234,14 +233,7 @@ async def main():
                 player_cache=player_cache,
             )
         )
-        asyncio.create_task(
-            check_duplicate_report(
-                report_queue=report_queue,
-                valid_report_queue=valid_report_queue,
-                skip=True,
-            )
-        )
-    asyncio.create_task(insert_batch(valid_report_queue=valid_report_queue))
+    asyncio.create_task(insert_batch(valid_report_queue=report_queue))
 
     while True:
         await asyncio.sleep(60)