Python Behave steps to func test Apache Kafka

kafka.feature

Scenario: send JSON to Kafka with Spark
    Given kafka topic
    When I run pipeline "output_kafka_spark.yml"
    Then I must see 3 messages from kafka

kafka_step.py

from behave import *
from hamcrest import *
from kafka import KafkaClient
from kafka import SimpleConsumer

@given('kafka topic')
def step_impl(context):
    c = context.djobi_config
    context.kafka_consumer = SimpleConsumer(client=KafkaClient(c["kafka.servers"]), topic=c.get_string("kafka.topic"), group="test")

@then('I must see {messages_count:d} messages from kafka')
def step_impl(context, messages_count):
    messages = context.kafka_consumer.get_messages(count=messages_count)
    assert_that(len(messages), equal_to(messages_count))