KAFKA-14569: Migrate Connect's integration test EmbeddedKafkaCluster from ZK to KRaft mode

[yashmayya]

• https://issues.apache.org/jira/browse/KAFKA-14569
• The existing EmbeddedConnectCluster (used in Connect integration tests) uses a backing EmbeddedKafkaCluster which internally also spins up an EmbeddedZookeeper.
• This patch migrates the EmbeddedKafkaCluster to run in KRaft mode, leveraging the existing KafkaClusterTestKit from core.
• Connect / MirrorMaker integration tests which setup a Kafka cluster with SASL_PLAINTEXT or SSL listeners needed to be updated to take into account the controller listeners as well and also update the authorizer used (from kafka.security.authorizer.AclAuthorizer to org.apache.kafka.metadata.authorizer.StandardAuthorizer, see KIP-801) since the old authorizer relied on ZooKeeper.
• The existing EmbeddedKafkaCluster had some logic to restart brokers and have them listening on the same ports as earlier (in order to verify Connect's functionality when its backing Kafka cluster goes down and then comes back up). This was refactored to move the responsibility of using a fixed port in the broker's listeners config to the tests themselves.
• Some changes to KafkaClusterTestKit in order to allow externally configuring listeners configurations and other minor improvements.

Committer Checklist (excluded from commit message)

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

[C0urante]

Thanks Yash! Made it part way through this one, will revisit and finish the round sometime this week (hopefully) or the next (if not).

[C0urante]

Any reason we're not preserving the startOnlyKafkaOnSamePorts method for this type of use case?

I see that this change is mentioned in the description, but don't see a rationale for it.

[yashmayya]

Sorry, my bad for not clarifying at the outset. There's a couple of reasons - one is that the previous implementation seemed fairly messy where we were checking the bound ports after broker startup and then modifying the broker config and using the static port in the listener configuration on broker restarts only if the user hadn't already configured listener configs. So, for instance, this startOnlyKafkaOnSamePorts would not work as expected if the user defines a SASL or SSL listener with port 0. The other reason is that the KafkaClusterTestKit being leveraged here would need some refactors to allow changing broker configs after it has already been instantiated - currently, it only supports customising broker configs in its builder. All in all, it just seems a lot cleaner to move the responsibility of using a fixed port in the listeners configuration to the clients of the embedded Kafka cluster in case they want to test functionality involving offline brokers.

[C0urante]

Fair enough, that works for me!
The one small change I'd like to see is that, in tests where we set listeners with a fixed non-zero port, we make sure to explicitly set the number of brokers to one (I know that this matches the current default, but we may change the default, and it'd be nice to make this test resilient against that).

[C0urante]

Actually, the more I think about it, the more brittle this seems. We're introducing a way to override the listeners property in the KafkaClusterTestKit class, but that only works for single-node clusters, and the use cases are fairly thin (IIUC it's not actually necessary to be able to override the listeners property for tests that involve changing the security protocols in use, which basically just leaves us with being able to reuse ports across restarts).

I'm still not completely against this (since if used responsibly, you can still do what you need to), but I wonder if it might be cleaner to do the refactoring in the KafkaClusterTestKit class instead, both to make it reusable and to allow for multi-node cases.

[yashmayya]

We're introducing a way to override the listeners property in the KafkaClusterTestKit class, but that only works for single-node clusters

The override ability is also currently used for setting up SASL / SSL listeners (which can work in multi-node cases using port 0).

IIUC it's not actually necessary to be able to override the listeners property for tests that involve changing the security protocols in use

I think you're suggesting adding an API to the KafkaClusterTestKit.Builder class to be able to conveniently setup SASL / SSL listeners instead of each test setting all the individual listener related configs?

[C0urante]

Correct me if I'm wrong, but there doesn't seem to be a case yet that requires overriding the listeners property that can't also be addressed by leaving that property as-is and only touching the listener.security.protocol.map property, except for reusing ports across restarts.

So yes, although it's possible to override listeners to set up SASL/SSL, this isn't required, and IMO it's actually cleaner if you do things by altering the security protocol map instead.

[C0urante]

I wonder if this has the potential to be a footgun. We're gating the default values of several properties on the presence of a single one; could it confuse people to set just the listeners property and then see failures because the previously-used defaults for the protocol map, controller listener, etc. weren't set?

[yashmayya]

Hm that's a good point and I did ponder over this one as well. The alternative would be to parse the user defined listener here and make sure that either the security protocol map, inter broker listener, controller listener are all set appropriately or else do so ourselves - this would involve making sure that the protocols, SASL mechanisms etc. all match up properly which seems potentially error prone (I tried a setup with broker listeners and controller listeners using different security protocols or mechanisms and I wasn't able to get it to work - not sure whether something like that is even currently supported). I'm assuming that users would likely want to set custom listeners in case they want non-plaintext listeners at which point it isn't much extra work to also setup the controller listener accordingly (and there's multiple examples on how to do so in various integration tests). What do you think?

[C0urante]

I think removing the if guard entirely and just doing putIfAbsent would be the most intuitive behavior. If someone still misconfigures their cluster, that's on them; we don't have to try to do any preflight checks here.

Then, if someone wants to enable SSL, they could just tweak the listener.security.protocol.map property and leave everything else as-is.

[C0urante]

Note: if we go that route, we should probably also tweak the logic for EmbeddedKafkaCluster::sslEnabled to handle when SSL is enabled by changing the listener security protocol map (instead of changing the listeners prop). One quick-and-dirty way to do this:

public boolean sslEnabled() {
    final String listeners = brokerConfig.getProperty(KafkaConfig.ListenersProp());
    final String protocolMap = brokerConfig.getProperty(KafkaConfig.ListenerSecurityProtocolMapProp());
    return (listeners != null && listeners.contains("SSL")) || (protocolMap != null && protocolMap.contains("SSL"));
}

[yashmayya]

Thanks for taking a look Chris!

[yashmayya]

Sorry, my bad for not clarifying at the outset. There's a couple of reasons - one is that the previous implementation seemed fairly messy where we were checking the bound ports after broker startup and then modifying the broker config and using the static port in the listener configuration on broker restarts only if the user hadn't already configured listener configs. So, for instance, this startOnlyKafkaOnSamePorts would not work as expected if the user defines a SASL or SSL listener with port 0. The other reason is that the KafkaClusterTestKit being leveraged here would need some refactors to allow changing broker configs after it has already been instantiated - currently, it only supports customising broker configs in its builder. All in all, it just seems a lot cleaner to move the responsibility of using a fixed port in the listeners configuration to the clients of the embedded Kafka cluster in case they want to test functionality involving offline brokers.

[yashmayya]

Hm that's a good point and I did ponder over this one as well. The alternative would be to parse the user defined listener here and make sure that either the security protocol map, inter broker listener, controller listener are all set appropriately or else do so ourselves - this would involve making sure that the protocols, SASL mechanisms etc. all match up properly which seems potentially error prone (I tried a setup with broker listeners and controller listeners using different security protocols or mechanisms and I wasn't able to get it to work - not sure whether something like that is even currently supported). I'm assuming that users would likely want to set custom listeners in case they want non-plaintext listeners at which point it isn't much extra work to also setup the controller listener accordingly (and there's multiple examples on how to do so in various integration tests). What do you think?

[C0urante]

Phew, all done! This finishes the first pass. Things are looking great, most of the comments should be easy to accommodate. One thing that'll require some thinking is whether we care about making the API for customizing listeners a bit friendlier, but we can always punt that for later if it's not worth the struggle for now.

Thanks for this migration Yash!

[github-actions]

This PR is being marked as stale since it has not had any activity in 90 days. If you would like to keep this PR alive, please ask a committer for review. If the PR has merge conflicts, please update it with the latest from trunk (or appropriate release branch)

If this PR is no longer valid or desired, please feel free to close it. If no activity occurrs in the next 30 days, it will be automatically closed.

[yashmayya]

Removing the stale label because I plan to re-visit this one.

[yashmayya]

@C0urante I've rebased this on the latest trunk, fixed the conflicts, and verified that all Connect and MM2 tests are passing locally. The two main remaining points of contention were configuring the listeners for SASL/SSL and for the test that required restarting Kafka brokers on the same port.

1. For the SASL/SSL listeners, I toyed with adding APIs to KafkaClusterTestKit in order to make configuring these listeners easier in tests. However, this led to an explosion of combinations in the builder because there are multiple configuration options for each unique security protocol and adding an API only for configuring the security protocol to be used by the listeners doesn't add much value. I opted to go with the more lightweight approach of configuring only the listener security protocol map in the tests (and retaining the listeners, inter-broker listener, controller listener etc. configured in the KafkaClusterTestKit) that you'd suggested earlier.
2. The restarting brokers on the same port bit proved to be a lot more vexatious though. I initially worked on refactoring the KafkaClusterTestKit class adding new methods to retain the startOnlyKafkaOnSamePorts functionality from the existing EmbeddedKafkaCluster. This involved instantiating new SharedServers for the BrokerServers with overridden listener configurations using the bound ports for the listener from the original BrokerServers. This ended up looking fairly ugly however and also runs the risk of port conflicts in parallel test runs (although this risk already exists). The other simpler approach was to add an API allowing users of KafkaClusterTestKit to configure static ports for all the broker listeners at the outset itself (i.e. in the original configuration used by the brokers). This would enable tests that require restarting Kafka and listening on the same ports to use multi-broker setups while being much cleaner than the previous approach. However, this will also lead to port conflict issues (even more likely probably). Overall, the port conflict issue makes it a little unappealing to add any such API to KafkaClusterTestKit as it would encourage further usage across other modules. Currently, there is only a single Connect test that makes use of such functionality and I think it probably makes more sense to simply use the static port in the overridden listeners configuration in just that test - I've added a comment clarifying that such an approach will only work for a single broker test and also explicitly configured the test to use a single broker like you'd suggested earlier.

[yashmayya]

testReplicationWithEmptyPartition was consistently timing out without this change, although I haven't yet gotten around to debugging why exactly that was the case..

[yashmayya]

This change was required to support Password type configs used in MirrorConnectorsIntegrationSSLTest

[mdedetrich]

@yashmayya Are you still working on this to get it over the finish line or is it okay for me to take over?

[yashmayya]

@mdedetrich apologies for the late response, I didn't get notified for your comment oddly enough. Please feel free to take over, thanks!

[C0urante]

I've published #16599, which is just a squash+rebase of this PR.