examples fixes

Author: idanasulin2706

examples/aiokafka/apache.py

@@ -21,7 +21,7 @@ async def create_producer(client_id):
     await producer.start()
     return producer
 
-async def send_messages_to_topics(producer, topics, producer_name, num_messages=50):
+async def send_messages_to_topics(producer, topics, producer_name, num_messages=5):
     """Send random JSON messages to specified Kafka topics"""
 
     successful = 0
@@ -36,7 +36,7 @@ async def send_messages_to_topics(producer, topics, producer_name, num_messages=
 
             # Send message to each topic
             for topic in topics:
-                result = await producer.send_and_wait(topic, message)
+                result = await producer.send(topic, message)
 
             successful += 1
 
@@ -55,30 +55,29 @@ async def main():
     try:
         # Create two separate producers
         producer1 = await create_producer('aiokafka-producer-1')
-        producer2 = await create_producer('aiokafka-producer-2')
+        # producer2 = await create_producer('aiokafka-producer-2')
 
         # First producer sends to test-topic and test-topic-1
         topics1 = ['test-topic', 'test-topic-1']
         await send_messages_to_topics(producer1, topics1, 'aiokafka-producer-1')
 
         # Second producer sends to test-topic-2 and test-topic-3
         topics2 = ['test-topic-2', 'test-topic-3']
-        await send_messages_to_topics(producer2, topics2, 'aiokafka-producer-2')
-        
+        # await send_messages_to_topics(producer2, topics2, 'aiokafka-producer-2')
+        # Sleep for 10 minutes at the end
+        print("Sleeping for 10 minutes...")
+        await asyncio.sleep(600)
+        print("Sleep completed")
     except Exception as e:
         print(f"Error: {e}")
     finally:
         if producer1:
             await producer1.stop()
             print("Producer 1 closed")
-        if producer2:
-            await producer2.stop()
-            print("Producer 2 closed")
+        # if producer2:
+        #     await producer2.stop()
+        #     print("Producer 2 closed")
 
-    # Sleep for 10 minutes at the end
-    print("Sleeping for 10 minutes...")
-    await asyncio.sleep(600)
-    print("Sleep completed")
 
 if __name__ == "__main__":
     # Run the async main function

examples/confluent_kafka/aiven.py

@@ -42,7 +42,7 @@ def create_producer(client_id):
         'ssl.ca.location': SSL_CAFILE,
         'ssl.certificate.location': SSL_CERTFILE,
         'ssl.key.location': SSL_KEYFILE,
-        'batch.num.messages': BATCH_SIZE,
+        'batch.size': BATCH_SIZE,
         'linger.ms': LINGER_MS
     })
 

examples/confluent_kafka/apache.py

@@ -56,9 +56,8 @@ def send_messages_to_topics(producer, topics, producer_name, num_messages=50):
             print(f"Failed to send message {i+1}: {e}")
 
         # Small delay between messages (optional)
-        time.sleep(0.01)
+        time.sleep(2)
 
-    producer.flush(timeout=30)
     print(f"\n{producer_name} Summary: {successful} successful, {failed} failed")
 
 def main():
@@ -67,23 +66,31 @@ def main():
     try:
         # Create two separate producers
         producer1 = create_producer('confluent-kafka-producer-1')
-        producer2 = create_producer('confluent-kafka-producer-2')
+        # producer2 = create_producer('confluent-kafka-producer-2')
 
         # First producer sends to test-topic and test-topic-1
         topics1 = ['test-topic', 'test-topic-1']
         send_messages_to_topics(producer1, topics1, 'confluent-kafka-producer-1')
 
         # Second producer sends to test-topic-2 and test-topic-3
         topics2 = ['test-topic-2', 'test-topic-3']
-        send_messages_to_topics(producer2, topics2, 'confluent-kafka-producer-2')
-        
+        # send_messages_to_topics(producer2, topics2, 'confluent-kafka-producer-2')
+        print("Sleeping...")
+        time.sleep(30)
+        print("Sleep completed")
     except Exception as e:
         print(f"Error: {e}")
 
-    # Sleep for 10 minutes at the end
-    print("Sleeping for 10 minutes...")
-    time.sleep(600)
-    print("Sleep completed")
+    # Explicitly set producers to None to force garbage collection
+    print("Setting producers to None to trigger Superstream cleanup...")
+    producer1 = None
+    # producer2 = None
+    
+    # Force garbage collection to trigger __del__ immediately
+    import gc
+    print("Forcing garbage collection...")
+    gc.collect()
+    print("Garbage collection completed")
 
 if __name__ == "__main__":
     main()

examples/confluent_kafka/confluent.py

@@ -40,7 +40,7 @@ def create_producer(client_id):
         'sasl.password': SASL_PASSWORD,
         'client.id': client_id,
         'linger.ms': LINGER_MS,
-        'batch.num.messages': BATCH_SIZE
+        'batch.size': BATCH_SIZE
     })
 
 def send_messages_to_topics(producer, topics, producer_name, num_messages=50):

examples/kafkapy/apache.py

@@ -6,7 +6,49 @@
 import time
 from kafka import KafkaProducer
 from kafka.errors import KafkaError
-from json_generator import generate_random_json
+
+import random
+import string
+from datetime import datetime
+
+def generate_random_json(min_size_kb=1):
+    """Generate a random JSON object of at least min_size_kb size"""
+    base_data = {
+        "timestamp": datetime.now().isoformat(),
+        "event_id": f"evt_{random.randint(100000, 999999)}",
+        "user_id": f"user_{random.randint(1000, 9999)}",
+        "session_id": f"session_{random.randint(10000, 99999)}",
+        "event_type": random.choice(["click", "view", "purchase", "login", "logout"]),
+        "device_type": random.choice(["mobile", "desktop", "tablet"]),
+        "os": random.choice(["Windows", "macOS", "Linux", "iOS", "Android"]),
+        "browser": random.choice(["Chrome", "Firefox", "Safari", "Edge"]),
+        "country": random.choice(["US", "UK", "DE", "FR", "JP", "BR", "IN"]),
+        "metrics": {
+            "load_time": round(random.uniform(0.1, 5.0), 3),
+            "response_time": round(random.uniform(0.01, 1.0), 3),
+            "cpu_usage": round(random.uniform(0, 100), 2),
+            "memory_usage": round(random.uniform(0, 100), 2)
+        }
+    }
+    
+    # Calculate current size
+    current_json = json.dumps(base_data)
+    current_size = len(current_json.encode('utf-8'))
+    target_size = min_size_kb * 1024
+    
+    # Add padding data if needed to reach target size
+    if current_size < target_size:
+        padding_size = target_size - current_size
+        # Generate random string data for padding
+        padding_data = {
+            "additional_data": {
+                f"field_{i}": ''.join(random.choices(string.ascii_letters + string.digits, k=50))
+                for i in range(padding_size // 50)
+            }
+        }
+        base_data.update(padding_data)
+    
+    return base_data
 
 def create_producer(client_id):
     """Create and configure Kafka producer"""
@@ -19,7 +61,7 @@ def create_producer(client_id):
         value_serializer=lambda v: json.dumps(v).encode('utf-8'),
     )
 
-def send_messages_to_topics(producer, topics, producer_name, num_messages=50):
+def send_messages_to_topics(producer, topics, producer_name, num_messages=5):
     """Send random JSON messages to specified Kafka topics"""
 
     successful = 0
@@ -57,30 +99,28 @@ def main():
     try:
         # Create two separate producers
         producer1 = create_producer('kafka-python-producer-1')
-        producer2 = create_producer('kafka-python-producer-2')
+        # producer2 = create_producer('kafka-python-producer-2')
 
         # First producer sends to test-topic and test-topic-1
         topics1 = ['test-topic', 'test-topic-1']
         send_messages_to_topics(producer1, topics1, 'kafka-python-producer-1')
-        
+        # Sleep for 10 minutes at the end
+        print("Sleeping for 10 minutes...")
+        time.sleep(600)
+        print("Sleep completed")
         # Second producer sends to test-topic-2 and test-topic-3
         topics2 = ['test-topic-2', 'test-topic-3']
-        send_messages_to_topics(producer2, topics2, 'kafka-python-producer-2')
+        # send_messages_to_topics(producer2, topics2, 'kafka-python-producer-2')
 
     except Exception as e:
         print(f"Error: {e}")
     finally:
         if producer1:
             producer1.close()
             print("Producer 1 closed")
-        if producer2:
-            producer2.close()
-            print("Producer 2 closed")
-    
-    # Sleep for 10 minutes at the end
-    print("Sleeping for 10 minutes...")
-    time.sleep(600)
-    print("Sleep completed")
+        # if producer2:
+        #     producer2.close()
+        #     print("Producer 2 closed")
 
 if __name__ == "__main__":
     main()