@Test public void getRyaInstance_invalidTopicName() { final String invalidTopic = "thisIsABadTopicName"; final Optional<String> ryaInstance = KafkaTopics.getRyaInstanceFromQueryChangeLog(invalidTopic); assertFalse( ryaInstance.isPresent() ); }
public static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic) { requireNonNull(changeLogTopic); if(!changeLogTopic.endsWith(QUERY_CHANGE_LOG_TOPIC_SUFFIX)) { return Optional.empty(); } final int endIndex = changeLogTopic.length() - QUERY_CHANGE_LOG_TOPIC_SUFFIX.length(); return Optional.of( changeLogTopic.substring(0, endIndex) ); }
KafkaTopics { public static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic) { requireNonNull(changeLogTopic); if(!changeLogTopic.endsWith(QUERY_CHANGE_LOG_TOPIC_SUFFIX)) { return Optional.empty(); } final int endIndex = changeLogTopic.length() - QUERY_CHANGE_LOG_TOPIC_SUFFIX.length(); return Optional.of( changeLogTopic.substring(0, endIndex) ); } }
KafkaTopics { public static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic) { requireNonNull(changeLogTopic); if(!changeLogTopic.endsWith(QUERY_CHANGE_LOG_TOPIC_SUFFIX)) { return Optional.empty(); } final int endIndex = changeLogTopic.length() - QUERY_CHANGE_LOG_TOPIC_SUFFIX.length(); return Optional.of( changeLogTopic.substring(0, endIndex) ); }  }
KafkaTopics { public static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic) { requireNonNull(changeLogTopic); if(!changeLogTopic.endsWith(QUERY_CHANGE_LOG_TOPIC_SUFFIX)) { return Optional.empty(); } final int endIndex = changeLogTopic.length() - QUERY_CHANGE_LOG_TOPIC_SUFFIX.length(); return Optional.of( changeLogTopic.substring(0, endIndex) ); }  static String queryChangeLogTopic(final String ryaInstance); static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic); static String statementsTopic(final String ryaInstance); static String queryResultsTopic(final String ryaInstance, final UUID queryId); static String getRyaInstanceFromQueryResultsTopic(final String queryResultsTopic); static UUID getQueryIdFromQueryResultsTopic(final String queryResultsTopic); static void createTopics(
            final String zookeeperServers,
            final Set<String> topicNames,
            final int partitions,
            final int replicationFactor,
            final Optional<Properties> topicProperties); }
KafkaTopics { public static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic) { requireNonNull(changeLogTopic); if(!changeLogTopic.endsWith(QUERY_CHANGE_LOG_TOPIC_SUFFIX)) { return Optional.empty(); } final int endIndex = changeLogTopic.length() - QUERY_CHANGE_LOG_TOPIC_SUFFIX.length(); return Optional.of( changeLogTopic.substring(0, endIndex) ); }  static String queryChangeLogTopic(final String ryaInstance); static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic); static String statementsTopic(final String ryaInstance); static String queryResultsTopic(final String ryaInstance, final UUID queryId); static String getRyaInstanceFromQueryResultsTopic(final String queryResultsTopic); static UUID getQueryIdFromQueryResultsTopic(final String queryResultsTopic); static void createTopics(
            final String zookeeperServers,
            final Set<String> topicNames,
            final int partitions,
            final int replicationFactor,
            final Optional<Properties> topicProperties);  }