@Test(expectedExceptions = IllegalArgumentException.class) public void testParseNoPrefix() { KafkaConnectionString.valueOf("hostname:666/topic"); }
public static KafkaConnectionString valueOf(String connectionString) throws IllegalArgumentException { if (connectionString == null) { badArg(null); } String str = connectionString.trim(); boolean isSecure = false; if (str.startsWith(PREFIX_SCHEME_KAFKA)) { str = str.substring(PREFIX_SCHEME_KAFKA.length()); isSecure = false; } else if (str.startsWith(PREFIX_SCHEME_SECURE_KAFKA)) { str = str.substring(PREFIX_SCHEME_SECURE_KAFKA.length()); isSecure = true; } else { badArg(connectionString); } int topicIndex = str.lastIndexOf("/"); if (topicIndex < 0) { badArg(connectionString); } String topicName = str.substring(topicIndex + 1).trim(); if (topicName.isEmpty()) { badArg(connectionString); } str = str.substring(0, topicIndex); List<KafkaBrokerAddress> brokers = parseBrokers(str); return new KafkaConnectionString(brokers, topicName, isSecure); }
KafkaConnectionString { public static KafkaConnectionString valueOf(String connectionString) throws IllegalArgumentException { if (connectionString == null) { badArg(null); } String str = connectionString.trim(); boolean isSecure = false; if (str.startsWith(PREFIX_SCHEME_KAFKA)) { str = str.substring(PREFIX_SCHEME_KAFKA.length()); isSecure = false; } else if (str.startsWith(PREFIX_SCHEME_SECURE_KAFKA)) { str = str.substring(PREFIX_SCHEME_SECURE_KAFKA.length()); isSecure = true; } else { badArg(connectionString); } int topicIndex = str.lastIndexOf("/"); if (topicIndex < 0) { badArg(connectionString); } String topicName = str.substring(topicIndex + 1).trim(); if (topicName.isEmpty()) { badArg(connectionString); } str = str.substring(0, topicIndex); List<KafkaBrokerAddress> brokers = parseBrokers(str); return new KafkaConnectionString(brokers, topicName, isSecure); } }
KafkaConnectionString { public static KafkaConnectionString valueOf(String connectionString) throws IllegalArgumentException { if (connectionString == null) { badArg(null); } String str = connectionString.trim(); boolean isSecure = false; if (str.startsWith(PREFIX_SCHEME_KAFKA)) { str = str.substring(PREFIX_SCHEME_KAFKA.length()); isSecure = false; } else if (str.startsWith(PREFIX_SCHEME_SECURE_KAFKA)) { str = str.substring(PREFIX_SCHEME_SECURE_KAFKA.length()); isSecure = true; } else { badArg(connectionString); } int topicIndex = str.lastIndexOf("/"); if (topicIndex < 0) { badArg(connectionString); } String topicName = str.substring(topicIndex + 1).trim(); if (topicName.isEmpty()) { badArg(connectionString); } str = str.substring(0, topicIndex); List<KafkaBrokerAddress> brokers = parseBrokers(str); return new KafkaConnectionString(brokers, topicName, isSecure); } KafkaConnectionString(List<KafkaBrokerAddress> brokers, String topicName, boolean isSecure); }
KafkaConnectionString { public static KafkaConnectionString valueOf(String connectionString) throws IllegalArgumentException { if (connectionString == null) { badArg(null); } String str = connectionString.trim(); boolean isSecure = false; if (str.startsWith(PREFIX_SCHEME_KAFKA)) { str = str.substring(PREFIX_SCHEME_KAFKA.length()); isSecure = false; } else if (str.startsWith(PREFIX_SCHEME_SECURE_KAFKA)) { str = str.substring(PREFIX_SCHEME_SECURE_KAFKA.length()); isSecure = true; } else { badArg(connectionString); } int topicIndex = str.lastIndexOf("/"); if (topicIndex < 0) { badArg(connectionString); } String topicName = str.substring(topicIndex + 1).trim(); if (topicName.isEmpty()) { badArg(connectionString); } str = str.substring(0, topicIndex); List<KafkaBrokerAddress> brokers = parseBrokers(str); return new KafkaConnectionString(brokers, topicName, isSecure); } KafkaConnectionString(List<KafkaBrokerAddress> brokers, String topicName, boolean isSecure); List<KafkaBrokerAddress> getBrokers(); String getTopicName(); boolean isSecure(); String getBrokerListString(); @Override String toString(); @Override boolean equals(Object o); @Override int hashCode(); static KafkaConnectionString valueOf(String connectionString); static List<KafkaBrokerAddress> parseBrokers(String brokersValue); }
KafkaConnectionString { public static KafkaConnectionString valueOf(String connectionString) throws IllegalArgumentException { if (connectionString == null) { badArg(null); } String str = connectionString.trim(); boolean isSecure = false; if (str.startsWith(PREFIX_SCHEME_KAFKA)) { str = str.substring(PREFIX_SCHEME_KAFKA.length()); isSecure = false; } else if (str.startsWith(PREFIX_SCHEME_SECURE_KAFKA)) { str = str.substring(PREFIX_SCHEME_SECURE_KAFKA.length()); isSecure = true; } else { badArg(connectionString); } int topicIndex = str.lastIndexOf("/"); if (topicIndex < 0) { badArg(connectionString); } String topicName = str.substring(topicIndex + 1).trim(); if (topicName.isEmpty()) { badArg(connectionString); } str = str.substring(0, topicIndex); List<KafkaBrokerAddress> brokers = parseBrokers(str); return new KafkaConnectionString(brokers, topicName, isSecure); } KafkaConnectionString(List<KafkaBrokerAddress> brokers, String topicName, boolean isSecure); List<KafkaBrokerAddress> getBrokers(); String getTopicName(); boolean isSecure(); String getBrokerListString(); @Override String toString(); @Override boolean equals(Object o); @Override int hashCode(); static KafkaConnectionString valueOf(String connectionString); static List<KafkaBrokerAddress> parseBrokers(String brokersValue); static final String BROKER_LIST_DELIMITER; }