KafkaMirrorMaker2Cluster extends KafkaConnectCluster { @SuppressWarnings({"checkstyle:CyclomaticComplexity", "checkstyle:NPathComplexity"}) @Override protected List<EnvVar> getEnvVars() { List<EnvVar> varList = super.getEnvVars(); final StringBuilder clusterAliases = new StringBuilder(); final StringBuilder clustersTrustedCerts = new StringBuilder(); final StringBuilder clustersTlsAuthCerts = new StringBuilder(); final StringBuilder clustersTlsAuthKeys = new StringBuilder(); final StringBuilder clustersSaslPasswordFiles = new StringBuilder(); boolean hasClusterOauthTrustedCerts = false; final StringBuilder clustersOauthClientSecrets = new StringBuilder(); final StringBuilder clustersOauthAccessTokens = new StringBuilder(); final StringBuilder clustersOauthRefreshTokens = new StringBuilder(); for (KafkaMirrorMaker2ClusterSpec mirrorMaker2Cluster : clusters) { String clusterAlias = mirrorMaker2Cluster.getAlias(); if (clusterAliases.length() > 0) { clusterAliases.append(";"); } clusterAliases.append(clusterAlias); getClusterTrustedCerts(clustersTrustedCerts, mirrorMaker2Cluster, clusterAlias); KafkaClientAuthentication authentication = mirrorMaker2Cluster.getAuthentication(); if (authentication != null) { if (authentication instanceof KafkaClientAuthenticationTls) { KafkaClientAuthenticationTls tlsAuth = (KafkaClientAuthenticationTls) authentication; if (tlsAuth.getCertificateAndKey() != null) { appendCluster(clustersTlsAuthCerts, clusterAlias, () -> tlsAuth.getCertificateAndKey().getSecretName() + "/" + tlsAuth.getCertificateAndKey().getCertificate()); appendCluster(clustersTlsAuthKeys, clusterAlias, () -> tlsAuth.getCertificateAndKey().getSecretName() + "/" + tlsAuth.getCertificateAndKey().getKey()); } } else if (authentication instanceof KafkaClientAuthenticationPlain) { KafkaClientAuthenticationPlain passwordAuth = (KafkaClientAuthenticationPlain) authentication; appendClusterPasswordSecretSource(clustersSaslPasswordFiles, clusterAlias, passwordAuth.getPasswordSecret()); } else if (authentication instanceof KafkaClientAuthenticationScramSha512) { KafkaClientAuthenticationScramSha512 passwordAuth = (KafkaClientAuthenticationScramSha512) authentication; appendClusterPasswordSecretSource(clustersSaslPasswordFiles, clusterAlias, passwordAuth.getPasswordSecret()); } else if (authentication instanceof KafkaClientAuthenticationOAuth) { KafkaClientAuthenticationOAuth oauth = (KafkaClientAuthenticationOAuth) authentication; if (oauth.getTlsTrustedCertificates() != null && !oauth.getTlsTrustedCertificates().isEmpty()) { hasClusterOauthTrustedCerts = true; } appendClusterOAuthSecretSource(clustersOauthClientSecrets, clusterAlias, oauth.getClientSecret()); appendClusterOAuthSecretSource(clustersOauthAccessTokens, clusterAlias, oauth.getAccessToken()); appendClusterOAuthSecretSource(clustersOauthRefreshTokens, clusterAlias, oauth.getRefreshToken()); } } } varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_CLUSTERS, clusterAliases.toString())); if (clustersTrustedCerts.length() > 0) { varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_TLS_CLUSTERS, "true")); varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_TRUSTED_CERTS_CLUSTERS, clustersTrustedCerts.toString())); } if (clustersTlsAuthCerts.length() > 0 || clustersTlsAuthKeys.length() > 0) { varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_TLS_AUTH_CLUSTERS, "true")); varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_TLS_AUTH_CERTS_CLUSTERS, clustersTlsAuthCerts.toString())); varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_TLS_AUTH_KEYS_CLUSTERS, clustersTlsAuthKeys.toString())); } if (clustersSaslPasswordFiles.length() > 0) { varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_SASL_PASSWORD_FILES_CLUSTERS, clustersSaslPasswordFiles.toString())); } if (hasClusterOauthTrustedCerts) { varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_OAUTH_TRUSTED_CERTS, "true")); } if (clustersOauthClientSecrets.length() > 0) { varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_OAUTH_CLIENT_SECRETS_CLUSTERS, clustersOauthClientSecrets.toString())); } if (clustersOauthAccessTokens.length() > 0) { varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_OAUTH_ACCESS_TOKENS_CLUSTERS, clustersOauthAccessTokens.toString())); } if (clustersOauthRefreshTokens.length() > 0) { varList.add(buildEnvVar(ENV_VAR_KAFKA_MIRRORMAKER_2_OAUTH_REFRESH_TOKENS_CLUSTERS, clustersOauthRefreshTokens.toString())); } if (javaSystemProperties != null) { varList.add(buildEnvVar(ENV_VAR_STRIMZI_JAVA_SYSTEM_PROPERTIES, ModelUtils.getJavaSystemPropertiesToString(javaSystemProperties))); } return varList; } private  KafkaMirrorMaker2Cluster(HasMetadata resource); static KafkaMirrorMaker2Cluster fromCrd(KafkaMirrorMaker2 kafkaMirrorMaker2, 
                                                   KafkaVersion.Lookup versions); @Override String getServiceAccountName();  }
@Test public void testEnvVars() { assertThat(kmm2.getEnvVars(), is(getExpectedEnvVars())); }