KafkaValidator extends AbstractProtocolValidator { @Override public void validate(final Connection connection, final DittoHeaders dittoHeaders, final ActorSystem actorSystem) { validateUriScheme(connection, dittoHeaders, ACCEPTED_SCHEMES, SECURE_SCHEMES, "Kafka 2.1.1"); validateSourceConfigs(connection, dittoHeaders); validateTargetConfigs(connection, dittoHeaders); validatePayloadMappings(connection, actorSystem, dittoHeaders); validateSpecificConfigs(connection, dittoHeaders); } private  KafkaValidator(); static KafkaValidator getInstance(); @Override ConnectionType type(); @Override void validate(final Connection connection, final DittoHeaders dittoHeaders, final ActorSystem actorSystem);  }
@Test public void testValidTargetAddress() { final DittoHeaders emptyDittoHeaders = DittoHeaders.empty(); underTest.validate(getConnectionWithTarget("events"), emptyDittoHeaders, actorSystem); underTest.validate(getConnectionWithTarget("ditto/{{thing:id}}"), emptyDittoHeaders, actorSystem); underTest.validate(getConnectionWithTarget("{{thing:namespace}}/{{thing:name}}"), emptyDittoHeaders, actorSystem); underTest.validate(getConnectionWithTarget("events#{{topic:full}}"), emptyDittoHeaders, actorSystem); underTest.validate(getConnectionWithTarget("ditto/{{header:x}}"), emptyDittoHeaders, actorSystem); }
@Test public void testValidBootstrapServers() { final DittoHeaders emptyDittoHeaders = DittoHeaders.empty(); underTest.validate(getConnectionWithBootstrapServers("foo:123"), emptyDittoHeaders, actorSystem); underTest.validate(getConnectionWithBootstrapServers("foo:123,bar:456"), emptyDittoHeaders, actorSystem); underTest.validate(getConnectionWithBootstrapServers("foo:123, bar:456 , baz:789"), emptyDittoHeaders, actorSystem); }