KafkaTopicExtractor { Optional<String> extractTopic(HttpServerRequest request) { final Logger requestLog = RequestLoggerFactory.getLogger(KafkaTopicExtractor.class, request); String topic = org.apache.commons.lang3.StringUtils.removeStart(request.uri(), streamingPath); if (StringUtils.isNotEmptyTrimmed(topic)) { return Optional.of(topic); } requestLog.warn("Extracted an empty string as topic from uri " + request.uri()); return Optional.empty(); } KafkaTopicExtractor(String streamingPath);   }
@Test public void emptyTopic(TestContext context) { TopicRequest request = new TopicRequest(PREFIX); context.assertFalse(extractor.extractTopic(request).isPresent()); }
@Test public void validTopic(TestContext context) { TopicRequest request = new TopicRequest(PREFIX + "my.topic.x"); final Optional<String> optTopic = extractor.extractTopic(request); context.assertTrue(optTopic.isPresent()); context.assertEquals("my.topic.x", optTopic.get()); TopicRequest request2 = new TopicRequest(PREFIX + "abc/def/ghi/123"); final Optional<String> optTopic2 = extractor.extractTopic(request2); context.assertTrue(optTopic2.isPresent()); context.assertEquals("abc/def/ghi/123", optTopic2.get()); }