@@ -0,0 +1,1082 @@
+import pytest%0A%0Afrom kafka import KafkaConsumer, KafkaProducer%0Afrom test.conftest import version%0Afrom test.testutil import random_string%0A%0A%0A@pytest.mark.skipif(not version(), reason=%22No KAFKA_VERSION set%22)%0Adef test_end_to_end(kafka_broker):%0A    connect_str = 'localhost:' + str(kafka_broker.port)%0A    producer = KafkaProducer(bootstrap_servers=connect_str,%0A                             max_block_ms=10000,%0A                             value_serializer=str.encode)%0A    consumer = KafkaConsumer(bootstrap_servers=connect_str,%0A                             consumer_timeout_ms=10000,%0A                             auto_offset_reset='earliest',%0A                             value_deserializer=bytes.decode)%0A%0A    topic = random_string(5)%0A%0A    for i in range(1000):%0A        producer.send(topic, 'msg %25d' %25 i)%0A    producer.flush()%0A    producer.close()%0A%0A    consumer.subscribe(%5Btopic%5D)%0A    msgs = set()%0A    for i in range(1000):%0A        try:%0A            msgs.add(next(consumer).value)%0A        except StopIteration:%0A            break%0A%0A    assert msgs == set(%5B'msg %25d' %25 i for i in range(1000)%5D)%0A

Add simple KafkaProducer -> KafkaConsumer integration test