def produce_and_consume(self, producer_version, consumer_version, group):
    self.producer = VerifiableProducer(self.test_context, self.num_producers, self.kafka, self.topic, throughput=self.producer_throughput, message_validator=is_int, version=KafkaVersion(producer_version))
    self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka, self.topic, consumer_timeout_ms=30000, message_validator=is_int, version=KafkaVersion(consumer_version))
    self.consumer.group_id = group
    self.run_produce_consume_validate((lambda : wait_until((lambda : (self.producer.each_produced_at_least(self.messages_per_producer) == True)), timeout_sec=120, backoff_sec=1, err_msg='Producer did not produce all messages in reasonable amount of time')))
def produce_and_consume(self, producer_version, consumer_version, group):
    self.producer = VerifiableProducer(self.test_context, self.num_producers, self.kafka, self.topic, throughput=self.producer_throughput, message_validator=is_int, version=KafkaVersion(producer_version))
    self.consumer = ConsoleConsumer(self.test_context, self.num_consumers, self.kafka, self.topic, consumer_timeout_ms=30000, message_validator=is_int, version=KafkaVersion(consumer_version))
    self.consumer.group_id = group
    self.run_produce_consume_validate((lambda : wait_until((lambda : (self.producer.each_produced_at_least(self.messages_per_producer) == True)), timeout_sec=120, backoff_sec=1, err_msg='Producer did not produce all messages in reasonable amount of time')))