Merge pull request confluentinc#4 from mapr/mapr-25766

Mapr 25766

Author: groghkov

confluent_kafka/kafkatest/verifiable_consumer.py

@@ -233,7 +233,7 @@ def to_dict (self):
     parser = argparse.ArgumentParser(description='Verifiable Python Consumer')
     parser.add_argument('--topic', action='append', type=str, required=True)
     parser.add_argument('--group-id', dest='group.id', required=True)
-    parser.add_argument('--broker-list', dest='bootstrap.servers', required=True)
+    parser.add_argument('--broker-list', dest='bootstrap.servers', required=False)
     parser.add_argument('--session-timeout', type=int, dest='session.timeout.ms', default=6000)
     parser.add_argument('--enable-autocommit', action='store_true', dest='enable.auto.commit', default=False)
     parser.add_argument('--max-messages', type=int, dest='max_messages', default=-1)

confluent_kafka/kafkatest/verifiable_producer.py

@@ -66,7 +66,7 @@ def dr_cb (self, err, msg):
     parser = argparse.ArgumentParser(description='Verifiable Python Producer')
     parser.add_argument('--topic', type=str, required=True)
     parser.add_argument('--throughput', type=int, default=0)
-    parser.add_argument('--broker-list', dest='bootstrap.servers', required=True)
+    parser.add_argument('--broker-list', dest='bootstrap.servers', required=False)
     parser.add_argument('--max-messages', type=int, dest='max_msgs', default=1000000) # avoid infinite
     parser.add_argument('--value-prefix', dest='value_prefix', type=str, default=None)
     parser.add_argument('--acks', type=int, dest='topic.request.required.acks', default=-1)

examples/consumer.py

@@ -23,18 +23,17 @@
 import sys
 
 if __name__ == '__main__':
-    if len(sys.argv) < 4:
-        sys.stderr.write('Usage: %s <bootstrap-brokers> <group> <topic1> <topic2> ..\n' % sys.argv[0])
+    if len(sys.argv) < 3:
+        sys.stderr.write('Usage: %s <group> <topic1> [<topic2> [..]]\n' % sys.argv[0])
         sys.exit(1)
 
-    broker = sys.argv[1]
-    group  = sys.argv[2]
-    topics = sys.argv[3:]
+    group  = sys.argv[1]
+    topics = sys.argv[2:]
 
     # Consumer configuration
     # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
-    conf = {'bootstrap.servers': broker, 'group.id': group, 'session.timeout.ms': 6000,
-            'default.topic.config': {'auto.offset.reset': 'smallest'}}
+    conf = {'group.id': group, 'session.timeout.ms': 6000,
+            'default.topic.config': {'auto.offset.reset': 'earliest'}}
 
 
     # Create Consumer instance

examples/integration_test.py

@@ -115,12 +115,11 @@ def verify_producer():
 
 def verify_producer_performance(with_dr_cb=True):
     """ Time how long it takes to produce and delivery X messages """
-    conf = {'bootstrap.servers': bootstrap_servers,
-            'error_cb': error_cb}
+    conf = {'error_cb': error_cb}
 
     p = confluent_kafka.Producer(**conf)
 
-    topic = 'test'
+    topic = '/test_stream:topic3'
     msgcnt = 1000000
     msgsize = 100
     msg_pattern = 'test.py performance'
@@ -141,9 +140,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
@@ -277,8 +276,7 @@ def verify_consumer():
 def verify_consumer_performance():
     """ Verify Consumer performance """
 
-    conf = {'bootstrap.servers': bootstrap_servers,
-            'group.id': uuid.uuid1(),
+    conf = {'group.id': uuid.uuid1(),
             'session.timeout.ms': 6000,
             'error_cb': error_cb,
             'default.topic.config': {
@@ -291,15 +289,13 @@ def my_on_assign (consumer, partitions):
         print('on_assign:', len(partitions), 'partitions:')
         for p in partitions:
             print(' %s [%d] @ %d' % (p.topic, p.partition, p.offset))
-        consumer.assign(partitions)
 
     def my_on_revoke (consumer, partitions):
         print('on_revoke:', len(partitions), 'partitions:')
         for p in 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(["/test_stream:topic3"], on_assign=my_on_assign, on_revoke=my_on_revoke)
 
     max_msgcnt = 1000000
     bytecnt = 0
@@ -362,12 +358,6 @@ def my_on_revoke (consumer, partitions):
     print('Using confluent_kafka module version %s (0x%x)' % confluent_kafka.version())
     print('Using librdkafka version %s (0x%x)' % confluent_kafka.libversion())
 
-    print('=' * 30, 'Verifying Producer', '=' * 30)
-    verify_producer()
-
-    print('=' * 30, 'Verifying Consumer', '=' * 30)
-    verify_consumer()
-
     print('=' * 30, 'Verifying Producer performance (with dr_cb)', '=' * 30)
     verify_producer_performance(with_dr_cb=True)
 

examples/producer.py

@@ -24,16 +24,16 @@
 import sys
 
 if __name__ == '__main__':
-    if len(sys.argv) != 3:
-        sys.stderr.write('Usage: %s <bootstrap-brokers> <topic>\n' % sys.argv[0])
+    if len(sys.argv) != 2:
+        sys.stderr.write('Usage: %s <topic>\n' % sys.argv[0])
         sys.exit(1)
 
-    broker = sys.argv[1]
-    topic  = sys.argv[2]
+    topic  = sys.argv[1]
 
     # Producer configuration
     # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
-    conf = {'bootstrap.servers': broker}
+    # Remove bootstrap.servers config
+    conf = {}
 
     # Create Producer instance
     p = Producer(**conf)
@@ -49,11 +49,23 @@ def delivery_callback (err, msg):
                              (msg.topic(), msg.partition()))
 
     # Read lines from stdin, produce each line to Kafka
-    for line in sys.stdin:
+    # Empty lines are ignored
+	while True:
+        try:
+            line = sys.stdin.readline()
+        except KeyboardInterrupt:
+            # CTRL + C
+	        break
+
+        if not line:
+            break
+
+        if line == '\n':
+            continue
+
         try:
             # Produce line (without newline)
             p.produce(topic, line.rstrip(), callback=delivery_callback)
-            
         except BufferError as e:
             sys.stderr.write('%% Local producer queue is full ' \
                              '(%d messages awaiting delivery): try again\n' %

setup.py

@@ -10,7 +10,7 @@
                              'confluent_kafka/src/Producer.c',
                              'confluent_kafka/src/Consumer.c'])
 
-setup(name='confluent-kafka',
+setup(name='mapr-streams-python',
       version='0.9.2',
       description='Confluent\'s Apache Kafka client for Python',
       author='Confluent Inc',

tests/test_Consumer.py

@@ -1,6 +1,28 @@
 #!/usr/bin/env python
 
 from confluent_kafka import Consumer, TopicPartition, KafkaError, KafkaException
+import pytest
+import subprocess
+import utils as u
+
+@pytest.fixture(scope="module", autouse=True)
+def create_stream(request):
+    u.new_stream('/stream', checked=True)
+    print("stream created")
+    def delete_stream():
+        u.delete_stream('/stream', checked=True)
+        print("stream deleted")
+    request.addfinalizer(delete_stream)
+    
+    
+@pytest.fixture(autouse=True)
+def resource_setup(request):
+    u.create_topic('/stream', 'topic1')
+    print("resource_setup")
+    def resource_teardown():
+        u.delete_topic('/stream', 'topic1')
+        print("resource_teardown")
+    request.addfinalizer(resource_teardown)
 
 
 def test_basic_api():
@@ -19,13 +41,13 @@ def dummy_commit_cb (err, partitions):
                    'session.timeout.ms': 1000, # Avoid close() blocking too long
                    'on_commit': dummy_commit_cb})
 
-    kc.subscribe(["test"])
+    kc.subscribe(["/stream:topic1"])
     kc.unsubscribe()
 
     def dummy_assign_revoke (consumer, partitions):
         pass
 
-    kc.subscribe(["test"], on_assign=dummy_assign_revoke, on_revoke=dummy_assign_revoke)
+    kc.subscribe(["/stream:topic1"], on_assign=dummy_assign_revoke, on_revoke=dummy_assign_revoke)
     kc.unsubscribe()
 
     msg = kc.poll(timeout=0.001)
@@ -36,10 +58,9 @@ def dummy_assign_revoke (consumer, partitions):
     else:
         print('OK: consumed message')
 
-    partitions = list(map(lambda p: TopicPartition("test", p), range(0,100,3)))
+    partitions = list(map(lambda p: TopicPartition("/stream:topic1", p), range(0,100,3)))
     kc.assign(partitions)
 
-    kc.unassign()
 
     kc.commit(async=True)
 
@@ -49,7 +70,12 @@ def dummy_assign_revoke (consumer, partitions):
         assert e.args[0].code() in (KafkaError._TIMED_OUT, KafkaError._NO_OFFSET)
 
     # Get current position, should all be invalid.
-    kc.position(partitions)
+    try:
+        kc.position(partitions)
+    except KafkaException as e:
+        print e
+        assert e.args[0].code() == KafkaError.TOPIC_EXCEPTION
+
     assert len([p for p in partitions if p.offset == -1001]) == len(partitions)
 
     try:

tests/test_Producer.py

@@ -1,12 +1,32 @@
 #!/usr/bin/env python
 
-from confluent_kafka import Producer, KafkaError, KafkaException
+from confluent_kafka import Producer, Consumer,  KafkaError, KafkaException, TopicPartition
+import mock
+import pytest
+import subprocess
+import utils as u
 
+@pytest.fixture(scope="module", autouse=True)
+def create_stream(request):
+    u.new_stream('/test_stream', checked=True)
+    print("stream created")
+    def delete_stream():
+        u.delete_stream('/test_stream', checked=True)
+        print("stream deleted")
+    request.addfinalizer(delete_stream)
+    
+    
+@pytest.fixture(autouse=True)
+def resource_setup(request):
+    u.create_topic('/test_stream', 'topic1')
+    print("resource_setup")
+    def resource_teardown():
+        u.delete_topic('/test_stream', 'topic1', checked=True)
+        print("resource_teardown")
+    request.addfinalizer(resource_teardown)
 
-def test_basic_api():
-    """ Basic API tests, these wont really do anything since there is no
-        broker configured. """
 
+def test_basic_api():
     try:
         p = Producer()
     except TypeError as e:
@@ -18,21 +38,58 @@ def error_cb (err):
 
     p = Producer({'socket.timeout.ms':10,
                   'error_cb': error_cb,
-                  'default.topic.config': {'message.timeout.ms': 10}})
+                  'default.topic.config': {'message.timeout.ms': 10, 'auto.offset.reset': 'earliest'}})
+
+    p.produce('/test_stream:topic1')
+    p.produce('/test_stream:topic1', value='somedata', key='a key')
+    try:
+        p.produce(None)
+    except TypeError as e:
+        assert str(e) == "argument 1 must be string, not None"
 
-    p.produce('mytopic')
-    p.produce('mytopic', value='somedata', key='a key')
 
     def on_delivery(err,msg):
         print('delivery', str)
-        # Since there is no broker, produced messages should time out.
-        assert err.code() == KafkaError._MSG_TIMED_OUT
 
-    p.produce(topic='another_topic', value='testing', partition=9,
+    p.produce(topic='/test_stream:topic1', value='testing', partition=9,
               callback=on_delivery)
 
     p.poll(0.001)
 
     p.flush()
 
 
+def test_producer_on_delivery():
+    p = Producer({'socket.timeout.ms':10,
+                  'default.topic.config': {'message.timeout.ms': 10, 'auto.offset.reset': 'earliest'}})
+    on_delivery = mock.Mock()
+    p.produce(topic='/test_stream:topic1', value='testing', partition=0,
+              callback=on_delivery)
+    p.flush()
+    assert on_delivery.called
+
+
+def test_producer_partition():
+    p = Producer({'socket.timeout.ms':10,
+                  'default.topic.config': {'message.timeout.ms': 10, 'auto.offset.reset': 'earliest'}})
+    p.produce(topic='/test_stream:topic2', value='testing', partition=0)
+    p.poll(1)
+    kc = Consumer({'group.id':'test', 'socket.timeout.ms':'100','enable.auto.commit': False,
+                   'session.timeout.ms': 1000, 'default.topic.config':{'auto.offset.reset': 'earliest'}})
+    kc.assign([TopicPartition("/test_stream:topic2", 0)])
+    msg = kc.poll()
+    assert  msg.value() == "testing"
+    kc.close()
+
+
+def test_producer_default_stream():
+    p = Producer({'socket.timeout.ms':10, 'streams.producer.default.stream': '/test_stream',
+                  'default.topic.config': {'message.timeout.ms': 10, 'auto.offset.reset': 'earliest'}})
+    p.produce(topic='topic1', value='TestDefaultStream')
+    p.poll(1)
+    kc = Consumer({'group.id':'test', 'socket.timeout.ms':'100','enable.auto.commit': False,
+                   'session.timeout.ms': 1000, 'default.topic.config':{'auto.offset.reset': 'earliest'}})
+    kc.assign([TopicPartition("/test_stream:topic1", 0)])
+    msg = kc.poll()
+    assert  msg.value() == "TestDefaultStream"
+    kc.close()