@cluster(num_nodes=12)
@parametrize(producer_version=str(DEV_BRANCH), consumer_version=str(DEV_BRANCH))
@parametrize(producer_version=str(LATEST_0_10), consumer_version=str(LATEST_0_10))
@parametrize(producer_version=str(LATEST_0_9), consumer_version=str(LATEST_0_9))
def test_compatibility(self, producer_version, consumer_version):
    " This tests performs the following checks:\n        The workload is a mix of 0.9.x, 0.10.x and 0.11.x producers and consumers\n        that produce to and consume from a DEV_BRANCH cluster\n        1. initially the topic is using message format 0.9.0\n        2. change the message format version for topic to 0.10.0 on the fly.\n        3. change the message format version for topic to 0.11.0 on the fly.\n        4. change the message format version for topic back to 0.10.0 on the fly (only if the client version is 0.11.0 or newer)\n        - The producers and consumers should not have any issue.\n\n        Note regarding step number 4. Downgrading the message format version is generally unsupported as it breaks\n        older clients. More concretely, if we downgrade a topic from 0.11.0 to 0.10.0 after it contains messages with\n        version 0.11.0, we will return the 0.11.0 messages without down conversion due to an optimisation in the\n        handling of fetch requests. This will break any consumer that doesn't support 0.11.0. So, in practice, step 4\n        is similar to step 2 and it didn't seem worth it to increase the cluster size to in order to add a step 5 that\n        would change the message format version for the topic back to 0.9.0.0.\n        "
    self.kafka = KafkaService(self.test_context, num_nodes=3, zk=self.zk, version=DEV_BRANCH, topics={self.topic: {'partitions': 3, 'replication-factor': 3, 'configs': {'min.insync.replicas': 2}}})
    self.kafka.start()
    self.logger.info('First format change to 0.9.0')
    self.kafka.alter_message_format(self.topic, str(LATEST_0_9))
    self.produce_and_consume(producer_version, consumer_version, 'group1')
    self.logger.info('Second format change to 0.10.0')
    self.kafka.alter_message_format(self.topic, str(LATEST_0_10))
    self.produce_and_consume(producer_version, consumer_version, 'group2')
    self.logger.info('Third format change to 0.11.0')
    self.kafka.alter_message_format(self.topic, str(LATEST_0_11))
    self.produce_and_consume(producer_version, consumer_version, 'group3')
    if ((producer_version == str(DEV_BRANCH)) and (consumer_version == str(DEV_BRANCH))):
        self.logger.info('Fourth format change back to 0.10.0')
        self.kafka.alter_message_format(self.topic, str(LATEST_0_10))
        self.produce_and_consume(producer_version, consumer_version, 'group4')
This tests performs the following checks:
The workload is a mix of 0.9.x, 0.10.x and 0.11.x producers and consumers
that produce to and consume from a DEV_BRANCH cluster
1. initially the topic is using message format 0.9.0
2. change the message format version for topic to 0.10.0 on the fly.
3. change the message format version for topic to 0.11.0 on the fly.
4. change the message format version for topic back to 0.10.0 on the fly (only if the client version is 0.11.0 or newer)
- The producers and consumers should not have any issue.

Note regarding step number 4. Downgrading the message format version is generally unsupported as it breaks
older clients. More concretely, if we downgrade a topic from 0.11.0 to 0.10.0 after it contains messages with
version 0.11.0, we will return the 0.11.0 messages without down conversion due to an optimisation in the
handling of fetch requests. This will break any consumer that doesn't support 0.11.0. So, in practice, step 4
is similar to step 2 and it didn't seem worth it to increase the cluster size to in order to add a step 5 that
would change the message format version for the topic back to 0.9.0.0.
@cluster(num_nodes=12)
@parametrize(producer_version=str(DEV_BRANCH), consumer_version=str(DEV_BRANCH))
@parametrize(producer_version=str(LATEST_0_10), consumer_version=str(LATEST_0_10))
@parametrize(producer_version=str(LATEST_0_9), consumer_version=str(LATEST_0_9))
def test_compatibility(self, producer_version, consumer_version):
    " This tests performs the following checks:\n        The workload is a mix of 0.9.x, 0.10.x and 0.11.x producers and consumers\n        that produce to and consume from a DEV_BRANCH cluster\n        1. initially the topic is using message format 0.9.0\n        2. change the message format version for topic to 0.10.0 on the fly.\n        3. change the message format version for topic to 0.11.0 on the fly.\n        4. change the message format version for topic back to 0.10.0 on the fly (only if the client version is 0.11.0 or newer)\n        - The producers and consumers should not have any issue.\n\n        Note regarding step number 4. Downgrading the message format version is generally unsupported as it breaks\n        older clients. More concretely, if we downgrade a topic from 0.11.0 to 0.10.0 after it contains messages with\n        version 0.11.0, we will return the 0.11.0 messages without down conversion due to an optimisation in the\n        handling of fetch requests. This will break any consumer that doesn't support 0.11.0. So, in practice, step 4\n        is similar to step 2 and it didn't seem worth it to increase the cluster size to in order to add a step 5 that\n        would change the message format version for the topic back to 0.9.0.0.\n        "
    self.kafka = KafkaService(self.test_context, num_nodes=3, zk=self.zk, version=DEV_BRANCH, topics={self.topic: {'partitions': 3, 'replication-factor': 3, 'configs': {'min.insync.replicas': 2}}})
    self.kafka.start()
    self.logger.info('First format change to 0.9.0')
    self.kafka.alter_message_format(self.topic, str(LATEST_0_9))
    self.produce_and_consume(producer_version, consumer_version, 'group1')
    self.logger.info('Second format change to 0.10.0')
    self.kafka.alter_message_format(self.topic, str(LATEST_0_10))
    self.produce_and_consume(producer_version, consumer_version, 'group2')
    self.logger.info('Third format change to 0.11.0')
    self.kafka.alter_message_format(self.topic, str(LATEST_0_11))
    self.produce_and_consume(producer_version, consumer_version, 'group3')
    if ((producer_version == str(DEV_BRANCH)) and (consumer_version == str(DEV_BRANCH))):
        self.logger.info('Fourth format change back to 0.10.0')
        self.kafka.alter_message_format(self.topic, str(LATEST_0_10))
        self.produce_and_consume(producer_version, consumer_version, 'group4')