<s> import logging <EOL> import os <EOL> from six . moves import xrange <EOL> from . import unittest <EOL> from kafka import ( <EOL> KafkaConsumer , MultiProcessConsumer , SimpleConsumer , create_message <EOL> ) <EOL> from kafka . consumer . base import MAX_FETCH_BUFFER_SIZE_BYTES <EOL> from kafka . errors import ConsumerFetchSizeTooSmall , OffsetOutOfRangeError <EOL> from kafka . structs import ProduceRequestPayload , TopicPartition <EOL> from test . fixtures import ZookeeperFixture , KafkaFixture <EOL> from test . testutil import ( <EOL> KafkaIntegrationTestCase , kafka_versions , random_string , Timer <EOL> ) <EOL> class TestConsumerIntegration ( KafkaIntegrationTestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> if not os . environ . get ( '<STR_LIT>' ) : <EOL> return <EOL> cls . zk = ZookeeperFixture . instance ( ) <EOL> chroot = random_string ( <NUM_LIT:10> ) <EOL> cls . server1 = KafkaFixture . instance ( <NUM_LIT:0> , cls . zk . host , cls . zk . port , <EOL> zk_chroot = chroot ) <EOL> cls . server2 = KafkaFixture . instance ( <NUM_LIT:1> , cls . zk . host , cls . zk . port , <EOL> zk_chroot = chroot ) <EOL> cls . server = cls . server1 <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> if not os . environ . get ( '<STR_LIT>' ) : <EOL> return <EOL> cls . server1 . close ( ) <EOL> cls . server2 . close ( ) <EOL> cls . zk . close ( ) <EOL> def send_messages ( self , partition , messages ) : <EOL> messages = [ create_message ( self . msg ( str ( msg ) ) ) for msg in messages ] <EOL> produce = ProduceRequestPayload ( self . topic , partition , messages = messages ) <EOL> resp , = self . client . send_produce_request ( [ produce ] ) <EOL> self . assertEqual ( resp . error , <NUM_LIT:0> ) <EOL> return [ x . value for x in messages ] <EOL> def assert_message_count ( self , messages , num_messages ) : <EOL> self . assertEqual ( len ( messages ) , num_messages ) <EOL> self . assertEqual ( len ( set ( messages ) ) , num_messages ) <EOL> def consumer ( self , ** kwargs ) : <EOL> if os . environ [ '<STR_LIT>' ] == "<STR_LIT>" : <EOL> kwargs [ '<STR_LIT>' ] = None <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> else : <EOL> kwargs . setdefault ( '<STR_LIT>' , None ) <EOL> kwargs . setdefault ( '<STR_LIT>' , False ) <EOL> consumer_class = kwargs . pop ( '<STR_LIT>' , SimpleConsumer ) <EOL> group = kwargs . pop ( '<STR_LIT>' , None ) <EOL> topic = kwargs . pop ( '<STR_LIT>' , self . topic ) <EOL> if consumer_class in [ SimpleConsumer , MultiProcessConsumer ] : <EOL> kwargs . setdefault ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> return consumer_class ( self . client , group , topic , ** kwargs ) <EOL> def kafka_consumer ( self , ** configs ) : <EOL> brokers = '<STR_LIT>' % ( self . server . host , self . server . port ) <EOL> consumer = KafkaConsumer ( self . topic , <EOL> bootstrap_servers = brokers , <EOL> ** configs ) <EOL> return consumer <EOL> def test_simple_consumer ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . consumer ( ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:200> ) <EOL> consumer . stop ( ) <EOL> def test_simple_consumer_smallest_offset_reset ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . consumer ( auto_offset_reset = '<STR_LIT>' ) <EOL> consumer . seek ( <NUM_LIT> , <NUM_LIT:2> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:200> ) <EOL> def test_simple_consumer_largest_offset_reset ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . consumer ( ) <EOL> consumer . seek ( <NUM_LIT> , <NUM_LIT:2> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:0> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:200> ) <EOL> def test_simple_consumer_no_reset ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . consumer ( auto_offset_reset = None ) <EOL> consumer . seek ( <NUM_LIT> , <NUM_LIT:2> ) <EOL> with self . assertRaises ( OffsetOutOfRangeError ) : <EOL> consumer . get_message ( ) <EOL> @ kafka_versions ( '<STR_LIT>' ) <EOL> def test_simple_consumer_load_initial_offsets ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . consumer ( group = '<STR_LIT>' ) <EOL> self . assertEqual ( consumer . offsets , { <NUM_LIT:0> : <NUM_LIT:0> , <NUM_LIT:1> : <NUM_LIT:0> } ) <EOL> consumer . offsets . update ( { <NUM_LIT:0> : <NUM_LIT> , <NUM_LIT:1> : <NUM_LIT> } ) <EOL> consumer . count_since_commit += <NUM_LIT:1> <EOL> consumer . commit ( ) <EOL> consumer = self . consumer ( group = '<STR_LIT>' , <EOL> auto_commit = False ) <EOL> self . assertEqual ( consumer . offsets , { <NUM_LIT:0> : <NUM_LIT> , <NUM_LIT:1> : <NUM_LIT> } ) <EOL> def test_simple_consumer__seek ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . consumer ( ) <EOL> consumer . seek ( - <NUM_LIT:10> , <NUM_LIT:2> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:10> ) <EOL> consumer . seek ( - <NUM_LIT> , <NUM_LIT:2> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT> ) <EOL> consumer . seek ( <NUM_LIT:100> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:0> ) <EOL> consumer . seek ( <NUM_LIT:100> , partition = <NUM_LIT:0> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:0> ) <EOL> consumer . seek ( <NUM_LIT> , partition = <NUM_LIT:1> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:0> ) <EOL> consumer . seek ( <NUM_LIT> , partition = <NUM_LIT:0> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:10> ) <EOL> consumer . seek ( <NUM_LIT:20> , partition = <NUM_LIT:1> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT> ) <EOL> consumer . seek ( <NUM_LIT:0> , partition = <NUM_LIT:1> ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:100> ) <EOL> consumer . stop ( ) <EOL> def test_simple_consumer_blocking ( self ) : <EOL> consumer = self . consumer ( ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( block = True , timeout = <NUM_LIT:1> ) <EOL> self . assert_message_count ( messages , <NUM_LIT:0> ) <EOL> self . assertGreaterEqual ( t . interval , <NUM_LIT:1> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:5> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:5> , <NUM_LIT:10> ) ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( count = <NUM_LIT:5> , block = True , timeout = <NUM_LIT:3> ) <EOL> self . assert_message_count ( messages , <NUM_LIT:5> ) <EOL> self . assertLess ( t . interval , <NUM_LIT:3> ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( count = <NUM_LIT:10> , block = True , timeout = <NUM_LIT:1> ) <EOL> self . assert_message_count ( messages , <NUM_LIT:5> ) <EOL> self . assertGreaterEqual ( t . interval , <NUM_LIT:1> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:3> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:3> , <NUM_LIT:5> ) ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( count = <NUM_LIT:10> , block = <NUM_LIT:1> , timeout = <NUM_LIT:1> ) <EOL> self . assert_message_count ( messages , <NUM_LIT:5> ) <EOL> self . assertLessEqual ( t . interval , <NUM_LIT:1> ) <EOL> consumer . stop ( ) <EOL> def test_simple_consumer_pending ( self ) : <EOL> consumer = self . consumer ( ) <EOL> self . assertEquals ( consumer . pending ( ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( consumer . pending ( partitions = [ <NUM_LIT:0> ] ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( consumer . pending ( partitions = [ <NUM_LIT:1> ] ) , <NUM_LIT:0> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:10> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:10> , <NUM_LIT:20> ) ) <EOL> consumer = self . consumer ( ) <EOL> self . assertEqual ( consumer . pending ( ) , <NUM_LIT:20> ) <EOL> self . assertEqual ( consumer . pending ( partitions = [ <NUM_LIT:0> ] ) , <NUM_LIT:10> ) <EOL> self . assertEqual ( consumer . pending ( partitions = [ <NUM_LIT:1> ] ) , <NUM_LIT:10> ) <EOL> consumer . seek ( - <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertEqual ( consumer . pending ( ) , <NUM_LIT:1> ) <EOL> pending_part1 = consumer . pending ( partitions = [ <NUM_LIT:0> ] ) <EOL> pending_part2 = consumer . pending ( partitions = [ <NUM_LIT:1> ] ) <EOL> self . assertEquals ( set ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) , set ( [ pending_part1 , pending_part2 ] ) ) <EOL> consumer . stop ( ) <EOL> @ unittest . skip ( '<STR_LIT>' ) <EOL> def test_multi_process_consumer ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . consumer ( consumer = MultiProcessConsumer ) <EOL> self . assert_message_count ( [ message for message in consumer ] , <NUM_LIT:200> ) <EOL> consumer . stop ( ) <EOL> @ unittest . skip ( '<STR_LIT>' ) <EOL> def test_multi_process_consumer_blocking ( self ) : <EOL> consumer = self . consumer ( consumer = MultiProcessConsumer ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( block = True , timeout = <NUM_LIT:1> ) <EOL> self . assert_message_count ( messages , <NUM_LIT:0> ) <EOL> self . assertGreaterEqual ( t . interval , <NUM_LIT:1> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:10> ) ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( count = <NUM_LIT:5> , block = True , timeout = <NUM_LIT:5> ) <EOL> self . assert_message_count ( messages , <NUM_LIT:5> ) <EOL> self . assertLessEqual ( t . interval , <NUM_LIT:1> ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( count = <NUM_LIT:10> , block = True , timeout = <NUM_LIT:1> ) <EOL> self . assert_message_count ( messages , <NUM_LIT:5> ) <EOL> self . assertGreaterEqual ( t . interval , <NUM_LIT:1> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:5> ) ) <EOL> with Timer ( ) as t : <EOL> messages = consumer . get_messages ( count = <NUM_LIT:10> , block = <NUM_LIT:1> , timeout = <NUM_LIT:1> ) <EOL> received_message_count = len ( messages ) <EOL> self . assertGreaterEqual ( received_message_count , <NUM_LIT:1> ) <EOL> self . assert_message_count ( messages , received_message_count ) <EOL> self . assertLessEqual ( t . interval , <NUM_LIT:1> ) <EOL> consumer . stop ( ) <EOL> @ unittest . skip ( '<STR_LIT>' ) <EOL> def test_multi_proc_pending ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:10> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:10> , <NUM_LIT:20> ) ) <EOL> consumer = MultiProcessConsumer ( self . client , None , self . topic , <EOL> auto_commit = False , iter_timeout = <NUM_LIT:0> ) <EOL> self . assertEqual ( consumer . pending ( ) , <NUM_LIT:20> ) <EOL> self . assertEqual ( consumer . pending ( partitions = [ <NUM_LIT:0> ] ) , <NUM_LIT:10> ) <EOL> self . assertEqual ( consumer . pending ( partitions = [ <NUM_LIT:1> ] ) , <NUM_LIT:10> ) <EOL> consumer . stop ( ) <EOL> @ unittest . skip ( '<STR_LIT>' ) <EOL> @ kafka_versions ( '<STR_LIT>' ) <EOL> def test_multi_process_consumer_load_initial_offsets ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:10> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:10> , <NUM_LIT:20> ) ) <EOL> consumer = self . consumer ( group = '<STR_LIT>' ) <EOL> self . assertEqual ( consumer . offsets , { <NUM_LIT:0> : <NUM_LIT:0> , <NUM_LIT:1> : <NUM_LIT:0> } ) <EOL> consumer . offsets . update ( { <NUM_LIT:0> : <NUM_LIT:5> , <NUM_LIT:1> : <NUM_LIT:15> } ) <EOL> consumer . count_since_commit += <NUM_LIT:1> <EOL> consumer . commit ( ) <EOL> consumer = self . consumer ( consumer = MultiProcessConsumer , <EOL> group = '<STR_LIT>' , <EOL> auto_commit = False ) <EOL> self . assertEqual ( consumer . offsets , { <NUM_LIT:0> : <NUM_LIT:5> , <NUM_LIT:1> : <NUM_LIT:15> } ) <EOL> def test_large_messages ( self ) : <EOL> small_messages = self . send_messages ( <NUM_LIT:0> , [ str ( x ) for x in range ( <NUM_LIT:10> ) ] ) <EOL> large_messages = self . send_messages ( <NUM_LIT:0> , [ random_string ( <NUM_LIT> ) for x in range ( <NUM_LIT:10> ) ] ) <EOL> consumer = self . consumer ( ) <EOL> expected_messages = set ( small_messages + large_messages ) <EOL> actual_messages = set ( [ x . message . value for x in consumer ] ) <EOL> self . assertEqual ( expected_messages , actual_messages ) <EOL> consumer . stop ( ) <EOL> def test_huge_messages ( self ) : <EOL> huge_message , = self . send_messages ( <NUM_LIT:0> , [ <EOL> create_message ( random_string ( MAX_FETCH_BUFFER_SIZE_BYTES + <NUM_LIT:10> ) ) , <EOL> ] ) <EOL> consumer = self . consumer ( ) <EOL> with self . assertRaises ( ConsumerFetchSizeTooSmall ) : <EOL> consumer . get_message ( False , <NUM_LIT:0.1> ) <EOL> consumer . stop ( ) <EOL> big_consumer = self . consumer ( <EOL> max_buffer_size = None , <EOL> partitions = [ <NUM_LIT:0> ] , <EOL> ) <EOL> big_consumer . seek ( - <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> message = big_consumer . get_message ( block = False , timeout = <NUM_LIT:10> ) <EOL> self . assertIsNotNone ( message ) <EOL> self . assertEqual ( message . message . value , huge_message ) <EOL> big_consumer . stop ( ) <EOL> @ kafka_versions ( '<STR_LIT>' ) <EOL> def test_offset_behavior__resuming_behavior ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer1 = self . consumer ( <EOL> group = '<STR_LIT>' , <EOL> auto_commit = True , <EOL> auto_commit_every_t = None , <EOL> auto_commit_every_n = <NUM_LIT:20> , <EOL> ) <EOL> output_msgs1 = [ consumer1 . get_message ( ) . message . value for _ in xrange ( <NUM_LIT> ) ] <EOL> self . assert_message_count ( output_msgs1 , <NUM_LIT> ) <EOL> consumer2 = self . consumer ( <EOL> group = '<STR_LIT>' , <EOL> auto_commit = True , <EOL> auto_commit_every_t = None , <EOL> auto_commit_every_n = <NUM_LIT:20> , <EOL> ) <EOL> self . assert_message_count ( [ message for message in consumer2 ] , <NUM_LIT:20> ) <EOL> consumer1 . stop ( ) <EOL> consumer2 . stop ( ) <EOL> @ unittest . skip ( '<STR_LIT>' ) <EOL> @ kafka_versions ( '<STR_LIT>' ) <EOL> def test_multi_process_offset_behavior__resuming_behavior ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer1 = self . consumer ( <EOL> consumer = MultiProcessConsumer , <EOL> group = '<STR_LIT>' , <EOL> auto_commit = True , <EOL> auto_commit_every_t = None , <EOL> auto_commit_every_n = <NUM_LIT:20> , <EOL> ) <EOL> output_msgs1 = [ ] <EOL> idx = <NUM_LIT:0> <EOL> for message in consumer1 : <EOL> output_msgs1 . append ( message . message . value ) <EOL> idx += <NUM_LIT:1> <EOL> if idx >= <NUM_LIT> : <EOL> break <EOL> self . assert_message_count ( output_msgs1 , <NUM_LIT> ) <EOL> consumer2 = self . consumer ( <EOL> consumer = MultiProcessConsumer , <EOL> group = '<STR_LIT>' , <EOL> auto_commit = True , <EOL> auto_commit_every_t = None , <EOL> auto_commit_every_n = <NUM_LIT:20> , <EOL> ) <EOL> self . assert_message_count ( [ message for message in consumer2 ] , <NUM_LIT:20> ) <EOL> consumer1 . stop ( ) <EOL> consumer2 . stop ( ) <EOL> def test_fetch_buffer_size ( self ) : <EOL> TEST_MESSAGE_SIZE = <NUM_LIT> <EOL> INIT_BUFFER_SIZE = <NUM_LIT> <EOL> MAX_BUFFER_SIZE = <NUM_LIT> <EOL> assert TEST_MESSAGE_SIZE > INIT_BUFFER_SIZE <EOL> assert TEST_MESSAGE_SIZE < MAX_BUFFER_SIZE <EOL> assert MAX_BUFFER_SIZE == <NUM_LIT:2> * INIT_BUFFER_SIZE <EOL> self . send_messages ( <NUM_LIT:0> , [ "<STR_LIT:x>" * <NUM_LIT> ] ) <EOL> self . send_messages ( <NUM_LIT:1> , [ "<STR_LIT:x>" * <NUM_LIT> ] ) <EOL> consumer = self . consumer ( buffer_size = <NUM_LIT> , max_buffer_size = <NUM_LIT> ) <EOL> messages = [ message for message in consumer ] <EOL> self . assertEqual ( len ( messages ) , <NUM_LIT:2> ) <EOL> def test_kafka_consumer ( self ) : <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer = self . kafka_consumer ( auto_offset_reset = '<STR_LIT>' ) <EOL> n = <NUM_LIT:0> <EOL> messages = { <NUM_LIT:0> : set ( ) , <NUM_LIT:1> : set ( ) } <EOL> for m in consumer : <EOL> logging . debug ( "<STR_LIT>" % repr ( m ) ) <EOL> n += <NUM_LIT:1> <EOL> messages [ m . partition ] . add ( m . offset ) <EOL> if n >= <NUM_LIT:200> : <EOL> break <EOL> self . assertEqual ( len ( messages [ <NUM_LIT:0> ] ) , <NUM_LIT:100> ) <EOL> self . assertEqual ( len ( messages [ <NUM_LIT:1> ] ) , <NUM_LIT:100> ) <EOL> def test_kafka_consumer__blocking ( self ) : <EOL> TIMEOUT_MS = <NUM_LIT> <EOL> consumer = self . kafka_consumer ( auto_offset_reset = '<STR_LIT>' , <EOL> consumer_timeout_ms = TIMEOUT_MS ) <EOL> consumer . unsubscribe ( ) <EOL> consumer . assign ( [ TopicPartition ( self . topic , <NUM_LIT:0> ) ] ) <EOL> with Timer ( ) as t : <EOL> with self . assertRaises ( StopIteration ) : <EOL> msg = next ( consumer ) <EOL> self . assertGreaterEqual ( t . interval , TIMEOUT_MS / <NUM_LIT> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:10> ) ) <EOL> messages = set ( ) <EOL> with Timer ( ) as t : <EOL> for i in range ( <NUM_LIT:5> ) : <EOL> msg = next ( consumer ) <EOL> messages . add ( ( msg . partition , msg . offset ) ) <EOL> self . assertEqual ( len ( messages ) , <NUM_LIT:5> ) <EOL> self . assertLess ( t . interval , TIMEOUT_MS / <NUM_LIT> ) <EOL> messages = set ( ) <EOL> with Timer ( ) as t : <EOL> with self . assertRaises ( StopIteration ) : <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> msg = next ( consumer ) <EOL> messages . add ( ( msg . partition , msg . offset ) ) <EOL> self . assertEqual ( len ( messages ) , <NUM_LIT:5> ) <EOL> self . assertGreaterEqual ( t . interval , TIMEOUT_MS / <NUM_LIT> ) <EOL> @ kafka_versions ( '<STR_LIT>' ) <EOL> def test_kafka_consumer__offset_commit_resume ( self ) : <EOL> GROUP_ID = random_string ( <NUM_LIT:10> ) <EOL> self . send_messages ( <NUM_LIT:0> , range ( <NUM_LIT:0> , <NUM_LIT:100> ) ) <EOL> self . send_messages ( <NUM_LIT:1> , range ( <NUM_LIT:100> , <NUM_LIT:200> ) ) <EOL> consumer1 = self . kafka_consumer ( <EOL> group_id = GROUP_ID , <EOL> enable_auto_commit = True , <EOL> auto_commit_interval_ms = <NUM_LIT:100> , <EOL> auto_offset_reset = '<STR_LIT>' , <EOL> ) <EOL> output_msgs1 = [ ] <EOL> for _ in xrange ( <NUM_LIT> ) : <EOL> m = next ( consumer1 ) <EOL> output_msgs1 . append ( m ) <EOL> self . assert_message_count ( output_msgs1 , <NUM_LIT> ) <EOL> consumer1 . close ( ) <EOL> consumer2 = self . kafka_consumer ( <EOL> group_id = GROUP_ID , <EOL> enable_auto_commit = True , <EOL> auto_commit_interval_ms = <NUM_LIT:100> , <EOL> auto_offset_reset = '<STR_LIT>' , <EOL> ) <EOL> output_msgs2 = [ ] <EOL> for _ in xrange ( <NUM_LIT:20> ) : <EOL> m = next ( consumer2 ) <EOL> output_msgs2 . append ( m ) <EOL> self . assert_message_count ( output_msgs2 , <NUM_LIT:20> ) <EOL> self . assertEqual ( len ( set ( output_msgs1 ) | set ( output_msgs2 ) ) , <NUM_LIT:200> ) </s>
