summaryrefslogtreecommitdiff
path: root/test/test_failover_integration.py
diff options
context:
space:
mode:
authorSpace <space@wibidata.com>2015-03-31 15:25:38 -0700
committerSpace <space@wibidata.com>2015-04-03 10:23:39 -0700
commit1c856e8400e1c4fe6dccd562fbcf4d1bde38755d (patch)
tree69e8a5142d8946adddc44590612a9968540d801f /test/test_failover_integration.py
parent9fd08119170b64c56ea024d12ef6b0e6482d778b (diff)
downloadkafka-python-1c856e8400e1c4fe6dccd562fbcf4d1bde38755d.tar.gz
Make external API consistently support python3 strings for topic.
Diffstat (limited to 'test/test_failover_integration.py')
-rw-r--r--test/test_failover_integration.py5
1 files changed, 3 insertions, 2 deletions
diff --git a/test/test_failover_integration.py b/test/test_failover_integration.py
index 15f0338..3be0189 100644
--- a/test/test_failover_integration.py
+++ b/test/test_failover_integration.py
@@ -8,6 +8,7 @@ from kafka import KafkaClient, SimpleConsumer
from kafka.common import TopicAndPartition, FailedPayloadsError, ConnectionError
from kafka.producer.base import Producer
from kafka.producer import KeyedProducer
+from kafka.util import kafka_bytestring
from test.fixtures import ZookeeperFixture, KafkaFixture
from test.testutil import (
@@ -147,7 +148,7 @@ class TestFailover(KafkaIntegrationTestCase):
key = random_string(3)
msg = random_string(10)
producer.send_messages(topic, key, msg)
- if producer.partitioners[topic].partition(key) == 0:
+ if producer.partitioners[kafka_bytestring(topic)].partition(key) == 0:
recovered = True
except (FailedPayloadsError, ConnectionError):
logging.debug("caught exception sending message -- will retry")
@@ -172,7 +173,7 @@ class TestFailover(KafkaIntegrationTestCase):
logging.debug('_send_random_message to %s:%d -- try %d success', topic, partition, j)
def _kill_leader(self, topic, partition):
- leader = self.client.topics_to_brokers[TopicAndPartition(topic, partition)]
+ leader = self.client.topics_to_brokers[TopicAndPartition(kafka_bytestring(topic), partition)]
broker = self.brokers[leader.nodeId]
broker.close()
return broker