static void validateConnectors(KafkaMirrorMaker2 kafkaMirrorMaker2)    {
        if (kafkaMirrorMaker2.getSpec() == null)    {
            throw new InvalidResourceException(".spec section is required for KafkaMirrorMaker2 resource");
        } else {
            if (kafkaMirrorMaker2.getSpec().getClusters() == null || kafkaMirrorMaker2.getSpec().getMirrors() == null)  {
                throw new InvalidResourceException(".spec.clusters and .spec.mirrors sections are required in KafkaMirrorMaker2 resource");
            } else {
                Set<String> existingClusterAliases = kafkaMirrorMaker2.getSpec().getClusters().stream().map(KafkaMirrorMaker2ClusterSpec::getAlias).collect(Collectors.toSet());
                Set<String> errorMessages = new HashSet<>();
                String connectCluster = kafkaMirrorMaker2.getSpec().getConnectCluster();

                for (KafkaMirrorMaker2MirrorSpec mirror : kafkaMirrorMaker2.getSpec().getMirrors())  {
                    if (mirror.getSourceCluster() == null)  {
                        errorMessages.add("Each MirrorMaker 2 mirror definition has to specify the source cluster alias");
                    } else if (!existingClusterAliases.contains(mirror.getSourceCluster())) {
                        errorMessages.add("Source cluster alias " + mirror.getSourceCluster() + " is used in a mirror definition, but cluster with this alias does not exist in cluster definitions");
                    }

                    if (mirror.getTargetCluster() == null)  {
                        errorMessages.add("Each MirrorMaker 2 mirror definition has to specify the target cluster alias");
                    } else if (!existingClusterAliases.contains(mirror.getTargetCluster())) {
                        errorMessages.add("Target cluster alias " + mirror.getTargetCluster() + " is used in a mirror definition, but cluster with this alias does not exist in cluster definitions");
                    }

                    if (!mirror.getTargetCluster().equals(connectCluster)) {
                        errorMessages.add("Connect cluster alias (currently set to " + connectCluster + ") has to be the same as the target cluster alias " + mirror.getTargetCluster());
                    }
                }

                if (!errorMessages.isEmpty())   {
                    throw new InvalidResourceException("KafkaMirrorMaker2 resource validation failed: " + errorMessages);
                }
            }
        }
    }
@Test
    public void testValidation()    {
        assertDoesNotThrow(() -> KafkaMirrorMaker2Connectors.validateConnectors(KMM2));
    }