Merge pull request #54 from Yelp/fix_merging_issues_with_142

88manpreet · web-flow · commit fd48b8c24cd3 · 2018-07-25T13:29:35.000-07:00
Fix upgrade issues with 1.4.2.post2 version
diff --git a/.dockerignore b/.dockerignore
@@ -0,0 +1,2 @@
+**/__pycache__
+**/*.pyc
diff --git a/Dockerfile b/Dockerfile
@@ -16,7 +16,7 @@ RUN apt-get update && apt-get install -y python2.7-dev \
     g++ \
     ca-certificates \
     python-pip \
-    python-tox 
+    python-tox
 
 # python-lz4 requires minium pypy version 5.8.0
 RUN wget https://bitbucket.org/pypy/pypy/downloads/pypy2-v5.8.0-linux64.tar.bz2
diff --git a/Makefile b/Makefile
@@ -16,16 +16,9 @@ itest:
 #   it passes the integration test locally if we disable ipv6 here
 	docker run --sysctl net.ipv6.conf.all.disable_ipv6=1 kafka_python_test /work/run_itest.sh
 
-clean:
-	rm -rf kafka-python.egg-info/ .tox/
-	find . -name '*.pyc' -delete
-	find . -name '__pycache__' -delete
-	docker rmi -f kafka_python_test
-
 docs:
 	tox -e docs
 
-.PHONY: docs test all
 
 FLAGS=
 KAFKA_VERSION=0.11.0.2
@@ -76,9 +69,9 @@ clean:
 	rm -rf docs/_build/
 	rm -rf cover
 	rm -rf dist
+	rm -rf kafka-python.egg-info/ .tox/
+	find . -name '*.pyc' -delete
+	find . -name '__pycache__' -delete
+	docker rmi -f kafka_python_test
 
-doc:
-	make -C docs html
-	@echo "open file://`pwd`/docs/_build/html/index.html"
-
-.PHONY: all test36 test27 test-local cov-local clean doc
+.PHONY: all test36 test27 test-local cov-local clean doc docs test
diff --git a/kafka/client.py b/kafka/client.py
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
@@ -638,7 +638,6 @@ def _poll_once(self, timeout_ms, max_records):
             # fetched records.
             if not partial:
                 self._fetcher.send_fetches()
-
             return records
 
         # Send any new fetches (won't resend pending fetches)
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
@@ -202,7 +202,6 @@ def _build_metadata_snapshot(self, subscription, cluster):
             metadata_snapshot[topic] = set(partitions)
         return metadata_snapshot
 
-
     def _lookup_assignor(self, name):
         for assignor in self.config['assignors']:
             if assignor.name == name:
diff --git a/kafka/protocol/commit.py b/kafka/protocol/commit.py
@@ -162,6 +162,7 @@ class OffsetFetchResponse_v3(Response):
         ('error_code', Int16)
     )
 
+
 class OffsetFetchRequest_v0(Request):
     API_KEY = 9
     API_VERSION = 0  # zookeeper-backed storage
diff --git a/kafka/protocol/message.py b/kafka/protocol/message.py
@@ -6,7 +6,6 @@
 from kafka.codec import (has_gzip, has_snappy, has_lz4,
                      gzip_decode, snappy_decode,
                      lz4_decode, lz4_decode_old_kafka)
-from ..util import crc32, WeakMethod
 from kafka.protocol.frame import KafkaBytes
 from kafka.protocol.struct import Struct
 from kafka.protocol.types import (
diff --git a/kafka/protocol/struct.py b/kafka/protocol/struct.py
@@ -7,8 +7,6 @@
 
 from kafka.util import WeakMethod
 
-from ..util import WeakMethod
-
 
 class Struct(AbstractType):
     SCHEMA = Schema()
diff --git a/run_itest.sh b/run_itest.sh
@@ -1,20 +1,19 @@
 #!/bin/bash -e
 
-
-export KAFKA_VERSION='0.9.0.1'
+export KAFKA_VERSION='0.10.1.1'
 ./build_integration.sh
 tox -e py27
 tox -e py35
 tox -e pypy
 
-export KAFKA_VERSION='0.10.0.0'
+export KAFKA_VERSION='0.11.0.2'
 ./build_integration.sh
 tox -e py27
 tox -e py35
 tox -e pypy
 
-export KAFKA_VERSION='0.10.1.1'
+export KAFKA_VERSION='1.0.1'
 ./build_integration.sh
-tox -e py27 
-tox -e py35 
-tox -e pypy 
+tox -e py27
+tox -e py35
+tox -e pypy
diff --git a/test/conftest.py b/test/conftest.py
@@ -3,7 +3,6 @@
 import inspect
 
 import pytest
-from decorator import decorate
 
 from test.fixtures import KafkaFixture, ZookeeperFixture
 from test.testutil import kafka_version, random_string
diff --git a/test/fixtures.py b/test/fixtures.py
@@ -296,9 +296,6 @@ def _create_zk_chroot(self):
                                          "kafka-python")
         env = self.kafka_run_class_env()
         proc = subprocess.Popen(args, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        # Fix bugs where integration tests hang in docker container by sending
-        # blank line to stdin
-        proc.communicate('')
 
         if proc.wait() != 0 or proc.returncode != 0:
             self.out("Failed to create Zookeeper chroot node")
diff --git a/test/test_client_integration.py b/test/test_client_integration.py
@@ -80,7 +80,6 @@ def test_send_produce_request_maintains_request_response_order(self):
     #   Offset Tests   #
     ####################
 
-    @kafka_versions('>=0.8.1')
     def test_commit_fetch_offsets(self):
         req = OffsetCommitRequestPayload(self.topic, 0, 42, 'metadata')
         (resp,) = self.client.send_offset_commit_request('group', [req])
@@ -92,3 +91,14 @@ def test_commit_fetch_offsets(self):
         self.assertEqual(resp.offset, 42)
         self.assertEqual(resp.metadata, '')  # Metadata isn't stored for now
 
+    @kafka_versions('>=0.9.0.0')
+    def test_commit_fetch_offsets_dual(self):
+        req = OffsetCommitRequestPayload(self.topic, 0, 42, 'metadata')
+        (resp,) = self.client.send_offset_commit_request_kafka('group', [req])
+        self.assertEqual(resp.error, 0)
+
+        (resp,) = self.client.send_offset_fetch_request_kafka('group', [req])
+        self.assertEqual(resp.error, 0)
+        self.assertEqual(resp.offset, 42)
+        # Metadata is stored in kafka
+        self.assertEqual(resp.metadata, 'metadata')
diff --git a/test/test_consumer.py b/test/test_consumer.py
@@ -3,7 +3,7 @@
 from mock import MagicMock, patch
 from . import unittest
 
-from kafka import SimpleConsumer, KafkaConsumer, MultiProcessConsumer
+from kafka import SimpleConsumer, KafkaConsumer, MultiProcessConsumer, OldKafkaConsumer
 from kafka.errors import (
     FailedPayloadsError, KafkaConfigurationError, NotLeaderForPartitionError,
     UnknownTopicOrPartitionError)
@@ -32,6 +32,9 @@ def test_subscription_copy(self):
         sub.add('fizz')
         assert consumer.subscription() == set(['foo'])
 
+    def test_broker_list_required(self):
+        with self.assertRaises(KafkaConfigurationError):
+            OldKafkaConsumer()
 
 class TestMultiProcessConsumer(unittest.TestCase):
     @unittest.skipIf(sys.platform.startswith('win'), 'test mocking fails on windows')
diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py
@@ -2,29 +2,32 @@
 import os
 import time
 
+import pytest
 from six.moves import xrange
 import six
 
 from . import unittest
 from kafka import (
-    KafkaConsumer, MultiProcessConsumer, SimpleConsumer, create_message,
+    KafkaConsumer, MultiProcessConsumer, OldKafkaConsumer, SimpleConsumer, create_message,
     create_gzip_message, KafkaProducer
 )
 from kafka.consumer.base import MAX_FETCH_BUFFER_SIZE_BYTES
 from kafka.errors import (
-    ConsumerFetchSizeTooSmall, OffsetOutOfRangeError, UnsupportedVersionError,
+    ConsumerFetchSizeTooSmall, OffsetOutOfRangeError, ConsumerTimeout, UnsupportedVersionError,
     KafkaTimeoutError
 )
 from kafka.structs import (
     ProduceRequestPayload, TopicPartition, OffsetAndTimestamp
 )
 
+from test.conftest import version
 from test.fixtures import ZookeeperFixture, KafkaFixture
 from test.testutil import (
     KafkaIntegrationTestCase, kafka_versions, random_string, Timer,
     send_messages
 )
 
+@pytest.mark.skipif(not version(), reason="No KAFKA_VERSION set")
 def test_kafka_consumer(simple_client, topic, kafka_consumer_factory):
     """Test KafkaConsumer
     """
@@ -761,3 +764,152 @@ def test_kafka_consumer_offsets_for_times_errors(self):
 
         with self.assertRaises(KafkaTimeoutError):
             consumer.offsets_for_times({bad_tp: 0})
+
+    def old_kafka_consumer(self, **configs):
+        brokers = '%s:%d' % (self.server.host, self.server.port)
+        consumer = OldKafkaConsumer(self.topic,
+                                    bootstrap_servers=brokers,
+                                    **configs)
+        return consumer
+ 
+    def test_old_kafka_consumer(self):
+        self.send_messages(0, range(0, 100))
+        self.send_messages(1, range(100, 200))
+ 
+        # Start a consumer
+        consumer = self.old_kafka_consumer(auto_offset_reset='smallest',
+                                        consumer_timeout_ms=5000)
+        n = 0
+        messages = {0: set(), 1: set()}
+        logging.debug("kafka consumer offsets: %s" % consumer.offsets())
+        for m in consumer:
+            logging.debug("Consumed message %s" % repr(m))
+            n += 1
+            messages[m.partition].add(m.offset)
+            if n == 200:
+                break
+ 
+        self.assertEqual(len(messages[0]), 100)
+        self.assertEqual(len(messages[1]), 100)
+ 
+    def test_old_kafka_consumer__blocking(self):
+        TIMEOUT_MS = 500
+        consumer = self.old_kafka_consumer(
+            auto_offset_reset='smallest',
+            consumer_timeout_ms=TIMEOUT_MS,
+        )
+ 
+        # Ask for 5 messages, nothing in queue, block 500ms
+        with Timer() as t:
+            with self.assertRaises(ConsumerTimeout):
+                msg = consumer.next()
+        self.assertGreaterEqual(t.interval, TIMEOUT_MS / 1000.0)
+ 
+        self.send_messages(0, range(0, 10))
+ 
+        # Ask for 5 messages, 10 in queue. Get 5 back, no blocking
+        messages = set()
+        with Timer() as t:
+            for i in range(5):
+                msg = consumer.next()
+                messages.add((msg.partition, msg.offset))
+        self.assertEqual(len(messages), 5)
+        self.assertLess(t.interval, TIMEOUT_MS / 1000.0)
+ 
+        # Ask for 10 messages, get 5 back, block 500ms
+        messages = set()
+        with Timer() as t:
+            with self.assertRaises(ConsumerTimeout):
+                for i in range(10):
+                    msg = consumer.next()
+                    messages.add((msg.partition, msg.offset))
+        self.assertEqual(len(messages), 5)
+        self.assertGreaterEqual(t.interval, TIMEOUT_MS / 1000.0)
+ 
+    @kafka_versions('=0.8.1')
+    def test_old_kafka_consumer__offset_commit_resume(self):
+        GROUP_ID = random_string(10).encode('utf-8')
+ 
+        self.send_messages(0, range(0, 100))
+        self.send_messages(1, range(100, 200))
+ 
+        # Start a consumer
+        consumer1 = self.old_kafka_consumer(
+            group_id=GROUP_ID,
+            auto_commit_enable=True,
+            auto_commit_interval_ms=None,
+            auto_commit_interval_messages=20,
+            auto_offset_reset='smallest',
+        )
+ 
+        # Grab the first 195 messages
+        output_msgs1 = []
+        for _ in xrange(195):
+            m = consumer1.next()
+            output_msgs1.append(m)
+            consumer1.task_done(m)
+        self.assert_message_count(output_msgs1, 195)
+ 
+        # The total offset across both partitions should be at 180
+        consumer2 = self.old_kafka_consumer(
+            group_id=GROUP_ID,
+            auto_commit_enable=True,
+            auto_commit_interval_ms=None,
+            auto_commit_interval_messages=20,
+            consumer_timeout_ms=100,
+            auto_offset_reset='smallest',
+        )
+ 
+        # 181-200
+        output_msgs2 = []
+        with self.assertRaises(ConsumerTimeout):
+            while True:
+                m = consumer2.next()
+                output_msgs2.append(m)
+        self.assert_message_count(output_msgs2, 20)
+        self.assertEqual(len(set(output_msgs1) & set(output_msgs2)), 15)
+ 
+    @kafka_versions("=0.9.0.0")
+    def test_old_kafka_consumer__offset_commit_resume_dual(self):
+        GROUP_ID = random_string(10).encode('utf-8')
+ 
+        self.send_messages(0, range(0, 100))
+        self.send_messages(1, range(100, 200))
+ 
+        # Start a consumer
+        consumer1 = self.old_kafka_consumer(
+            group_id=GROUP_ID,
+            auto_commit_enable=True,
+            auto_commit_interval_ms=None,
+            auto_commit_interval_messages=20,
+            auto_offset_reset='smallest',
+            offset_storage='kafka',
+        )
+ 
+        # Grab the first 195 messages
+        output_msgs1 = []
+        for _ in xrange(195):
+            m = consumer1.next()
+            output_msgs1.append(m)
+            consumer1.task_done(m)
+        self.assert_message_count(output_msgs1, 195)
+ 
+        # The total offset across both partitions should be at 180
+        consumer2 = self.old_kafka_consumer(
+            group_id=GROUP_ID,
+            auto_commit_enable=True,
+            auto_commit_interval_ms=None,
+            auto_commit_interval_messages=20,
+            consumer_timeout_ms=100,
+            auto_offset_reset='smallest',
+            offset_storage='dual',
+        )
+ 
+        # 181-200
+        output_msgs2 = []
+        with self.assertRaises(ConsumerTimeout):
+            while True:
+                m = consumer2.next()
+                output_msgs2.append(m)
+        self.assert_message_count(output_msgs2, 20)
+        self.assertEqual(len(set(output_msgs1) & set(output_msgs2)), 15)
diff --git a/test/test_producer_integration.py b/test/test_producer_integration.py
@@ -2,6 +2,7 @@
 import time
 import uuid
 
+import pytest
 from six.moves import range
 
 from kafka import (
@@ -14,6 +15,7 @@
 from kafka.producer.base import Producer
 from kafka.structs import FetchRequestPayload, ProduceRequestPayload
 
+from test.conftest import version
 from test.fixtures import ZookeeperFixture, KafkaFixture
 from test.testutil import KafkaIntegrationTestCase, kafka_versions, current_offset
 
@@ -39,6 +41,7 @@ def assert_produce_response(resp, initial_offset):
     assert resp[0].error == 0
     assert resp[0].offset == initial_offset
 
+@pytest.mark.skipif(not version(), reason="No KAFKA_VERSION set")
 def test_produce_many_simple(simple_client, topic):
     """Test multiple produces using the SimpleClient
     """
diff --git a/tox.ini b/tox.ini

Original file line number	Diff line number	Diff line change
`@@ -162,6 +162,7 @@ class OffsetFetchResponse_v3(Response):`
`162`	`162`	`('error_code', Int16)`
`163`	`163`	`)`
`164`	`164`
	`165`	`+`
`165`	`166`	`class OffsetFetchRequest_v0(Request):`
`166`	`167`	`API_KEY = 9`
`167`	`168`	`API_VERSION = 0 # zookeeper-backed storage`