KafkaDestination { public String getDestinationURI() { return String.format(_isSecure ? DESTINATION_URI_SSL_FORMAT : DESTINATION_URI_FORMAT, _zkAddress, _topicName); } KafkaDestination(String zkAddress, String topicName, boolean isSecure); static KafkaDestination parse(String uri); String getZkAddress(); String getTopicName(); boolean isSecure(); String getDestinationURI(); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); static final String SCHEME_KAFKA; static final String SCHEME_SECURE_KAFKA; static final String DESTINATION_URI_FORMAT; static final String DESTINATION_URI_SSL_FORMAT; }
@Test public void testDestinationFormatting() { String zkAddress = "localhost:12913/kafka-datastream"; String topicName = "testtopic_test"; KafkaDestination destination = new KafkaDestination(zkAddress, topicName, false); Assert.assertEquals(destination.getDestinationURI(), "kafka: destination = new KafkaDestination(zkAddress, topicName, true); Assert.assertEquals(destination.getDestinationURI(), "kafkassl: zkAddress = "localhost:12913"; destination = new KafkaDestination(zkAddress, topicName, true); Assert.assertEquals(destination.getDestinationURI(), "kafkassl: }