KafkaAuthenticationSpecificConfig implements KafkaSpecificConfig { @Override public void validateOrThrow(final Connection connection, final DittoHeaders dittoHeaders) { if (!isValid(connection)) { final String message = MessageFormat.format( "The connection configuration contains an invalid value for SASL mechanisms: <{0}>. Allowed " + "mechanisms are: <{1}>", getSaslMechanismOrDefault(connection), SASL_MECHANISMS_WITH_LOGIN_MODULE.keySet()); throw ConnectionConfigurationInvalidException.newBuilder(message) .dittoHeaders(dittoHeaders) .build(); } } private  KafkaAuthenticationSpecificConfig(); static KafkaAuthenticationSpecificConfig getInstance(); @Override boolean isApplicable(final Connection connection); @Override void validateOrThrow(final Connection connection, final DittoHeaders dittoHeaders); @Override boolean isValid(final Connection connection); @Override void apply(final HashMap<String, Object> producerProperties, final Connection connection);  }
@Test public void shouldValidateIfContainsNoSaslMechanism() { underTest.validateOrThrow(getConnectionWithSaslMechanism(null), HEADERS); }
@Test public void shouldValidateIfContainsSupportedSaslMechanism() { underTest.validateOrThrow(getConnectionWithSaslMechanism(KNOWN_PLAIN_SASL_MECHANISM), HEADERS); underTest.validateOrThrow(getConnectionWithSaslMechanism(KNOWN_SHA256_SASL_MECHANISM), HEADERS); underTest.validateOrThrow(getConnectionWithSaslMechanism(KNOWN_SHA512_SASL_MECHANISM), HEADERS); }