diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/test_consumer_integration.py | 6 | ||||
| -rw-r--r-- | test/test_producer_integration.py | 4 |
2 files changed, 8 insertions, 2 deletions
diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py index a2b8f70..d6fd41c 100644 --- a/test/test_consumer_integration.py +++ b/test/test_consumer_integration.py @@ -17,6 +17,7 @@ from kafka.errors import ( ConsumerFetchSizeTooSmall, OffsetOutOfRangeError, UnsupportedVersionError, KafkaTimeoutError, UnsupportedCodecError ) +from kafka.protocol.message import PartialMessage from kafka.structs import ( ProduceRequestPayload, TopicPartition, OffsetAndTimestamp ) @@ -249,6 +250,8 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): consumer.stop() + @pytest.mark.skipif(env_kafka_version() >= (2, 0), + reason="SimpleConsumer blocking does not handle PartialMessage change in kafka 2.0+") def test_simple_consumer_blocking(self): consumer = self.consumer() @@ -414,7 +417,8 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): consumer = self.consumer(max_buffer_size=60000) expected_messages = set(small_messages + large_messages) - actual_messages = set([ x.message.value for x in consumer ]) + actual_messages = set([x.message.value for x in consumer + if not isinstance(x.message, PartialMessage)]) self.assertEqual(expected_messages, actual_messages) consumer.stop() diff --git a/test/test_producer_integration.py b/test/test_producer_integration.py index e0939a6..8f32cf8 100644 --- a/test/test_producer_integration.py +++ b/test/test_producer_integration.py @@ -13,6 +13,7 @@ from kafka import ( from kafka.codec import has_snappy from kafka.errors import UnknownTopicOrPartitionError, LeaderNotAvailableError from kafka.producer.base import Producer +from kafka.protocol.message import PartialMessage from kafka.structs import FetchRequestPayload, ProduceRequestPayload from test.fixtures import ZookeeperFixture, KafkaFixture @@ -521,7 +522,8 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase): self.assertEqual(resp.error, 0) self.assertEqual(resp.partition, partition) - messages = [ x.message.value for x in resp.messages ] + messages = [ x.message.value for x in resp.messages + if not isinstance(x.message, PartialMessage) ] self.assertEqual(messages, expected_messages) self.assertEqual(resp.highwaterMark, start_offset+len(expected_messages)) |