[FLINK-2386] [kafka] Add new Kafka Consumer for Flink

This closes #1055
apache · Aug 26, 2015 · 940a7c8 · 940a7c8
1 parent 3cdbb80
commit 940a7c8
Show file tree

Hide file tree

Showing 42 changed files with 5,183 additions and 2,133 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/util/NetUtils.java b/flink-core/src/main/java/org/apache/flink/util/NetUtils.java
@@ -49,7 +49,7 @@ public static String getHostnameFromFQDN(String fqdn) {
 	 *
 	 * See: http://stackoverflow.com/questions/2345063/java-common-way-to-validate-and-convert-hostport-to-inetsocketaddress
 	 */
-	public static void ensureCorrectHostnamePort(String hostPort) {
+	public static URL getHostnamePort(String hostPort) {
 		try {
 			URL u = new URL("http://"+hostPort);
 			if(u.getHost() == null) {
@@ -58,6 +58,7 @@ public static void ensureCorrectHostnamePort(String hostPort) {
 			if(u.getPort() == -1) {
 				throw new IllegalArgumentException("The given host:port ('"+hostPort+"') doesn't contain a valid port");
 			}
+			return u;
 		} catch (MalformedURLException e) {
 			throw new IllegalArgumentException("The given host:port ('"+hostPort+"') is invalid", e);
 		}

diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/pom.xml b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/pom.xml
@@ -92,10 +92,24 @@ under the License.
 			</exclusions>
 		</dependency>
 
+		<dependency>
+			<groupId>com.google.guava</groupId>
+			<artifactId>guava</artifactId>
+			<version>${guava.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.curator</groupId>
 			<artifactId>curator-test</artifactId>
 			<version>2.7.1</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-test-utils</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
 		</dependency>
 
 	</dependencies>