@Test(expectedExceptions = Exception.class) public void testUriMissingAuthority() { KafkaDestination.parse("kafka: }
public static KafkaDestination parse(String uri) { Validate.isTrue(uri.startsWith(SCHEME_KAFKA) || uri.startsWith(SCHEME_SECURE_KAFKA), "Invalid scheme in URI: " + uri); try { uri = URIUtil.decode(uri); } catch (Exception e) { throw new DatastreamRuntimeException("Failed to decode Kafka destination URI: " + uri, e); } URI u = URI.create(uri); String scheme = u.getScheme(); String zkAddress = u.getAuthority(); String path = u.getPath(); String topicName; int lastSlash = path.lastIndexOf("/"); if (lastSlash > 0) { zkAddress += path.substring(0, lastSlash); topicName = path.substring(lastSlash + 1); } else { topicName = path.substring(1); } for (String hostInfo : zkAddress.split(",")) { long portNum = URI.create(SCHEME_KAFKA + ": Validate.isTrue(portNum != -1, "Missing port number in URI: " + uri); } Validate.notBlank(zkAddress, "Missing zkAddress in URI: " + uri); Validate.notBlank(topicName, "Missing topic name in URI: " + uri); boolean isSecure = scheme.equals(SCHEME_SECURE_KAFKA); return new KafkaDestination(zkAddress, topicName, isSecure); }
KafkaDestination { public static KafkaDestination parse(String uri) { Validate.isTrue(uri.startsWith(SCHEME_KAFKA) || uri.startsWith(SCHEME_SECURE_KAFKA), "Invalid scheme in URI: " + uri); try { uri = URIUtil.decode(uri); } catch (Exception e) { throw new DatastreamRuntimeException("Failed to decode Kafka destination URI: " + uri, e); } URI u = URI.create(uri); String scheme = u.getScheme(); String zkAddress = u.getAuthority(); String path = u.getPath(); String topicName; int lastSlash = path.lastIndexOf("/"); if (lastSlash > 0) { zkAddress += path.substring(0, lastSlash); topicName = path.substring(lastSlash + 1); } else { topicName = path.substring(1); } for (String hostInfo : zkAddress.split(",")) { long portNum = URI.create(SCHEME_KAFKA + ": Validate.isTrue(portNum != -1, "Missing port number in URI: " + uri); } Validate.notBlank(zkAddress, "Missing zkAddress in URI: " + uri); Validate.notBlank(topicName, "Missing topic name in URI: " + uri); boolean isSecure = scheme.equals(SCHEME_SECURE_KAFKA); return new KafkaDestination(zkAddress, topicName, isSecure); } }
KafkaDestination { public static KafkaDestination parse(String uri) { Validate.isTrue(uri.startsWith(SCHEME_KAFKA) || uri.startsWith(SCHEME_SECURE_KAFKA), "Invalid scheme in URI: " + uri); try { uri = URIUtil.decode(uri); } catch (Exception e) { throw new DatastreamRuntimeException("Failed to decode Kafka destination URI: " + uri, e); } URI u = URI.create(uri); String scheme = u.getScheme(); String zkAddress = u.getAuthority(); String path = u.getPath(); String topicName; int lastSlash = path.lastIndexOf("/"); if (lastSlash > 0) { zkAddress += path.substring(0, lastSlash); topicName = path.substring(lastSlash + 1); } else { topicName = path.substring(1); } for (String hostInfo : zkAddress.split(",")) { long portNum = URI.create(SCHEME_KAFKA + ": Validate.isTrue(portNum != -1, "Missing port number in URI: " + uri); } Validate.notBlank(zkAddress, "Missing zkAddress in URI: " + uri); Validate.notBlank(topicName, "Missing topic name in URI: " + uri); boolean isSecure = scheme.equals(SCHEME_SECURE_KAFKA); return new KafkaDestination(zkAddress, topicName, isSecure); } KafkaDestination(String zkAddress, String topicName, boolean isSecure); }
KafkaDestination { public static KafkaDestination parse(String uri) { Validate.isTrue(uri.startsWith(SCHEME_KAFKA) || uri.startsWith(SCHEME_SECURE_KAFKA), "Invalid scheme in URI: " + uri); try { uri = URIUtil.decode(uri); } catch (Exception e) { throw new DatastreamRuntimeException("Failed to decode Kafka destination URI: " + uri, e); } URI u = URI.create(uri); String scheme = u.getScheme(); String zkAddress = u.getAuthority(); String path = u.getPath(); String topicName; int lastSlash = path.lastIndexOf("/"); if (lastSlash > 0) { zkAddress += path.substring(0, lastSlash); topicName = path.substring(lastSlash + 1); } else { topicName = path.substring(1); } for (String hostInfo : zkAddress.split(",")) { long portNum = URI.create(SCHEME_KAFKA + ": Validate.isTrue(portNum != -1, "Missing port number in URI: " + uri); } Validate.notBlank(zkAddress, "Missing zkAddress in URI: " + uri); Validate.notBlank(topicName, "Missing topic name in URI: " + uri); boolean isSecure = scheme.equals(SCHEME_SECURE_KAFKA); return new KafkaDestination(zkAddress, topicName, isSecure); } 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(); }
KafkaDestination { public static KafkaDestination parse(String uri) { Validate.isTrue(uri.startsWith(SCHEME_KAFKA) || uri.startsWith(SCHEME_SECURE_KAFKA), "Invalid scheme in URI: " + uri); try { uri = URIUtil.decode(uri); } catch (Exception e) { throw new DatastreamRuntimeException("Failed to decode Kafka destination URI: " + uri, e); } URI u = URI.create(uri); String scheme = u.getScheme(); String zkAddress = u.getAuthority(); String path = u.getPath(); String topicName; int lastSlash = path.lastIndexOf("/"); if (lastSlash > 0) { zkAddress += path.substring(0, lastSlash); topicName = path.substring(lastSlash + 1); } else { topicName = path.substring(1); } for (String hostInfo : zkAddress.split(",")) { long portNum = URI.create(SCHEME_KAFKA + ": Validate.isTrue(portNum != -1, "Missing port number in URI: " + uri); } Validate.notBlank(zkAddress, "Missing zkAddress in URI: " + uri); Validate.notBlank(topicName, "Missing topic name in URI: " + uri); boolean isSecure = scheme.equals(SCHEME_SECURE_KAFKA); return new KafkaDestination(zkAddress, topicName, isSecure); } 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; }