MirrorMaker2: Check if bootstrap servers are the same for different cluster aliases

[varada-sunanda-ibm]

Type of change

Select the type of your PR

• Bugfix
• Refactoring

Description

After the changes we made in the PR #9923, the current condition prevents users from independently managing the connections between Kafka Connect and Kafka, as well as between the MM2 connectors and Kafka.
Specifically, we’re using two distinct KafkaUsers, each assigned with minimal set of ACLs that are required.

Our proposed solution is to modify the condition to permit scenarios where the connectCluster bootstrapServer and the mirror.targetCluster bootstrapServer are identical.

Checklist

Please go through this checklist and make sure all applicable tasks have been done

• Write tests
• Make sure all tests pass
• Update documentation
• Check RBAC rights for Kubernetes / OpenShift roles
• Try your changes from Pod inside your Kubernetes and OpenShift cluster, not just locally
• Reference relevant issue(s) and close them after merging
• Update CHANGELOG.md
• Supply screenshots for visual changes, such as Grafana dashboards

[varada-sunanda-ibm]

Slack Discussion : https://cloud-native.slack.com/archives/CMH3Q3SNP/p1725430611080659

[ppatierno]

I had a pass with no additional comments to the ones already there. I will have another one when comments are addressed.

[katheris]

LGTM

[scholzj]

This looks super complex and hard to read. Why don't you pass the list of clusters here, find the one for the first alias, find the one for the second alias and then compare them?

[scholzj]

This should be aligned according to the edit / end methods. Same applies to other two tests below.

Suggested change

	.editSpec()
	.withConnectCluster("source")
	.addToClusters(new KafkaMirrorMaker2ClusterSpecBuilder()
	.withAlias("third")
	.withBootstrapServers("source:9092")
	.build())
	.editMirror(0)
	.withTargetCluster("third")
	.endMirror()
	.endSpec()
	.editSpec()
	.withConnectCluster("source")
	.addToClusters(new KafkaMirrorMaker2ClusterSpecBuilder()
	.withAlias("third")
	.withBootstrapServers("source:9092")
	.build())
	.editMirror(0)
	.withTargetCluster("third")
	.endMirror()
	.endSpec()

[scholzj]

If you insist on doing it this comúplicated way despite my comments, maybe you can add at least some comment/JavaDoc to what it does and how.

[varada-sunanda-ibm]

Apologies @scholzj, I missed that earlier comment. Did you mean something like this?

private static boolean hasMatchingBootstrapServers(List<KafkaMirrorMaker2ClusterSpec> clusterList, String connectClusterAlias, String targetClusterAlias) {
    // Find the cluster for the connectClusterAlias
    String connectClusterBootstrap = clusterList.stream()
            .filter(cluster -> connectClusterAlias.equals(cluster.getAlias()))
            .map(KafkaMirrorMaker2ClusterSpec::getBootstrapServers)
            .findFirst()
            .orElse(null);

    // Find the cluster for the targetClusterAlias
    String targetClusterBootstrap = clusterList.stream()
            .filter(cluster -> targetClusterAlias.equals(cluster.getAlias()))
            .map(KafkaMirrorMaker2ClusterSpec::getBootstrapServers)
            .findFirst()
            .orElse(null);

    // Return true if both are found and have matching bootstrap servers
    return connectClusterBootstrap != null && connectClusterBootstrap.equals(targetClusterBootstrap);
}

Let me know if this is what you were suggesting.

[scholzj]

Yes, this is what I would have used.

[varada-sunanda-ibm]

Sure. I have pushed in the latest changes. Please can you re-review.

[scholzj]

/azp run regression

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[scholzj]

LGTM. Thanks.

[ppatierno]

LGTM. Thanks!