MINOR: Support Raft-based metadata quorums in system tests

[rondagostino]

We need to be able to run system tests with Raft-based metadata quorums -- both co-located brokers and controllers as well as remote controllers -- in addition to the ZooKepeer-based mode we run today. This PR adds this capability to KafkaService in a backwards-compatible manner as follows.

If no changes are made to existing system tests then they function as they always do -- they instantiate ZooKeeper, and Kafka will use ZooKeeper. A good test of this PR is therefore to run a full system test suite with no actual test changes and make sure everything runs as expected.

If we want to use a Raft-based metadata quorum we can do so by introducing a metadata_quorum argument to the test method and using @matrix to set it to the quorums we want to use for the various runs of the test. We then also have to skip creating a ZooKeeperService when the quorum is Raft-based.

For example, we would do the following:

from ducktape.mark import matrix
from kafkatest.services.kafka import KafkaService, quorum

    def __init__(self, test_context):
        super(TestVerifiableProducer, self).__init__(test_context)
        self.zk = ZookeeperService(test_context, num_nodes=1) if quorum.for_test(test_context) == quorum.zk else None
        self.kafka = KafkaService(test_context, num_nodes=1, zk=self.zk,
                                  topics={"topic": {"partitions": 1, "replication-factor": 1}})

    def setUp(self):
        if self.zk:
            self.zk.start()
        self.kafka.start()

    @cluster(num_nodes=3)
    @matrix(producer_version=[str(DEV_BRANCH)], metadata_quorum=quorum.all)
    def test_simple_run(self, producer_version=DEV_BRANCH, metadata_quorum=quorum.zk):
        # the rest of the test logic remains unchanged

The above will end up running 3 separate tests: one with ZooKeeper, one with a co-located Raft-based controller, and once with a remote Raft-based controller.

If we want to set security protocols we could do this:

    def setUp(self):
        if self.zk:
            self.zk.start()
        # don't start Kafka here because we haven't configured security at this point

    @cluster(num_nodes=3)
    @matrix(producer_version=[str(DEV_BRANCH)], security_protocol=['PLAINTEXT', 'SSL'], metadata_quorum=quorum.all)
    @cluster(num_nodes=4)
    @matrix(producer_version=[str(DEV_BRANCH)], security_protocol=['SASL_SSL'], sasl_mechanism=['PLAIN', 'GSSAPI'],
            metadata_quorum=quorum.all)
    def test_simple_run(self, producer_version, security_protocol = 'PLAINTEXT', sasl_mechanism='PLAIN',
                        metadata_quorum=quorum.zk):
        self.kafka.security_protocol = security_protocol
        self.kafka.client_sasl_mechanism = sasl_mechanism
        self.kafka.interbroker_security_protocol = security_protocol
        self.kafka.interbroker_sasl_mechanism = sasl_mechanism
        if self.kafka.quorum_info.using_raft:
            controller_quorum = self.kafka.controller_quorum
            controller_quorum.controller_security_protocol = security_protocol
            controller_quorum.controller_sasl_mechanism = sasl_mechanism
            controller_quorum.intercontroller_security_protocol = security_protocol
            controller_quorum.intercontroller_sasl_mechanism = sasl_mechanism
        # now we can start Kafka
        self.kafka.start()
        # the rest of the test logic remains unchanged

This PR does not update any tests -- those will come later after all the KIP-500 code is merged.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[cmccabe]

LGTM

[rondagostino]

System test results:

SESSION REPORT (ALL TESTS)
ducktape version: 0.8.1
session_id:       2021-02-10--001
run time:         509 minutes 34.017 seconds
tests run:        725
passed:           425
failed:           103
ignored:          197

http://testing.confluent.io/confluent-kafka-branch-builder-system-test-results/?prefix=2021-02-10--001.1612998000--rondagostino--kip500_system_tests--b16fc75f6/report.html

Compared to a system test run from the day before:

SESSION REPORT (ALL TESTS)
ducktape version: 0.8.1
session_id:       2021-02-09--001
run time:         388 minutes 22.578 seconds
tests run:        729
passed:           453
failed:           79
ignored:          197

http://testing.confluent.io/confluent-kafka-branch-builder-system-test-results/?prefix=2021-02-09--001.1612926625--urbandan--KIP-635_GetOffsetShell--8e69e1dc5/report.html

There are lots more failures, but I looked and the failures are not due to the system tests not running. For example, an error message is java.lang.Exception: UnsupportedVersionException: MetadataRequest versions older than 4 don't support the allowAutoTopicCreation field.

So this PR is not affecting the running of the tests, which is all that matters.