integration_test.py: require broker to be configured, allow topic name conf

Author: edenhill

README.md

@@ -115,7 +115,7 @@ Tests
 
 **Run integration tests:**
 
-    $ examples/integration_test.py <kafka-broker>
+    $ examples/integration_test.py <kafka-broker> [<test-topic>]
 
 **WARNING**: These tests require an active Kafka cluster and will make use of a topic named 'test'.
 

examples/integration_test.py

@@ -33,8 +33,10 @@
     with_progress = False
 
 # Kafka bootstrap server(s)
-bootstrap_servers = 'localhost'
+bootstrap_servers = None
 
+# Topic to use
+topic = 'test'
 
 # global variable to be set by stats_cb call back function
 good_stats_cb_result = False
@@ -90,22 +92,22 @@ def verify_producer():
     print('producer at %s' % p)
 
     # Produce some messages
-    p.produce('test', 'Hello Python!')
-    p.produce('test', key='Just a key')
-    p.produce('test', partition=1, value='Strictly for partition 1',
+    p.produce(topic, 'Hello Python!')
+    p.produce(topic, key='Just a key')
+    p.produce(topic, partition=1, value='Strictly for partition 1',
               key='mykey')
 
     # Produce more messages, now with delivery report callbacks in various forms.
     mydr = MyTestDr()
-    p.produce('test', value='This one has a dr callback',
+    p.produce(topic, value='This one has a dr callback',
               callback=mydr.delivery)
-    p.produce('test', value='This one has a lambda',
+    p.produce(topic, value='This one has a lambda',
               callback=lambda err, msg: MyTestDr._delivery(err, msg))
-    p.produce('test', value='This one has neither')
+    p.produce(topic, value='This one has neither')
 
     # Produce even more messages
     for i in range(0, 10):
-        p.produce('test', value='Message #%d' % i, key=str(i),
+        p.produce(topic, value='Message #%d' % i, key=str(i),
                   callback=mydr.delivery)
         p.poll(0)
 
@@ -123,7 +125,6 @@ def verify_producer_performance(with_dr_cb=True):
 
     p = confluent_kafka.Producer(**conf)
 
-    topic = 'test'
     msgcnt = 1000000
     msgsize = 100
     msg_pattern = 'test.py performance'
@@ -144,9 +145,9 @@ def verify_producer_performance(with_dr_cb=True):
     for i in range(0, msgcnt):
         try:
             if with_dr_cb:
-                p.produce('test', value=msg_payload, callback=dr.delivery)
+                p.produce(topic, value=msg_payload, callback=dr.delivery)
             else:
-                p.produce('test', value=msg_payload)
+                p.produce(topic, value=msg_payload)
         except BufferError as e:
             # Local queue is full (slow broker connection?)
             msgs_backpressure += 1
@@ -224,7 +225,7 @@ def verify_consumer():
     c = confluent_kafka.Consumer(**conf)
 
     # Subscribe to a list of topics
-    c.subscribe(["test"])
+    c.subscribe([topic])
 
     max_msgcnt = 100
     msgcnt = 0
@@ -270,7 +271,7 @@ def verify_consumer():
 
     # Start a new client and get the committed offsets
     c = confluent_kafka.Consumer(**conf)
-    offsets = c.committed(list(map(lambda p: confluent_kafka.TopicPartition("test", p), range(0,3))))
+    offsets = c.committed(list(map(lambda p: confluent_kafka.TopicPartition(topic, p), range(0,3))))
     for tp in offsets:
         print(tp)
 
@@ -304,7 +305,7 @@ def my_on_revoke (consumer, partitions):
             print(' %s [%d] @ %d' % (p.topic, p.partition, p.offset))
         consumer.unassign()
 
-    c.subscribe(["test"], on_assign=my_on_assign, on_revoke=my_on_revoke)
+    c.subscribe([topic], on_assign=my_on_assign, on_revoke=my_on_revoke)
 
     max_msgcnt = 1000000
     bytecnt = 0
@@ -364,10 +365,11 @@ def verify_stats_cb():
     def stats_cb(stats_json_str):
         global good_stats_cb_result
         stats_json = json.loads(stats_json_str)
-        if 'test' in stats_json['topics']:
-            app_offset = stats_json['topics']['test']['partitions']['0']['app_offset']
+        if topic in stats_json['topics']:
+            app_offset = stats_json['topics'][topic]['partitions']['0']['app_offset']
             if app_offset > 0:
-                print("# app_offset stats for topic test partition 0: %d" % app_offset)
+                print("# app_offset stats for topic %s partition 0: %d" % \
+                      (topic, app_offset))
                 good_stats_cb_result = True
 
     conf = {'bootstrap.servers': bootstrap_servers,
@@ -381,7 +383,7 @@ def stats_cb(stats_json_str):
             }}
 
     c = confluent_kafka.Consumer(**conf)
-    c.subscribe(["test"])
+    c.subscribe([topic])
 
     max_msgcnt = 1000000
     bytecnt = 0
@@ -439,6 +441,11 @@ def stats_cb(stats_json_str):
 
     if len(sys.argv) > 1:
         bootstrap_servers = sys.argv[1]
+        if len(sys.argv) > 2:
+            topic = sys.argv[2]
+    else:
+        print('Usage: %s <broker> [<topic>]' % sys.argv[0])
+        sys.exit(1)
 
     print('Using confluent_kafka module version %s (0x%x)' % confluent_kafka.version())
     print('Using librdkafka version %s (0x%x)' % confluent_kafka.libversion())