You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/08/27 13:25:52 UTC
[35/51] [abbrv] flink git commit: [FLINK-2386] [kafka connector]
Remove copied Kafka code again. Implemented our own topic metadata retrieval.
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/PartitionValidatingMapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/PartitionValidatingMapper.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/PartitionValidatingMapper.java
deleted file mode 100644
index c59e779..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/PartitionValidatingMapper.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.testutils;
-
-import org.apache.flink.api.common.functions.MapFunction;
-
-import java.util.HashSet;
-import java.util.Set;
-
-
-public class PartitionValidatingMapper implements MapFunction<Integer, Integer> {
-
- private static final long serialVersionUID = 1088381231244959088L;
-
- /* the partitions from which this function received data */
- private final Set<Integer> myPartitions = new HashSet<>();
-
- private final int numPartitions;
- private final int maxPartitions;
-
- public PartitionValidatingMapper(int numPartitions, int maxPartitions) {
- this.numPartitions = numPartitions;
- this.maxPartitions = maxPartitions;
- }
-
- @Override
- public Integer map(Integer value) throws Exception {
- // validate that the partitioning is identical
- int partition = value % numPartitions;
- myPartitions.add(partition);
- if (myPartitions.size() > maxPartitions) {
- throw new Exception("Error: Elements from too many different partitions: " + myPartitions
- + ". Expect elements only from " + maxPartitions + " partitions");
- }
- return value;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/SuccessException.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/SuccessException.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/SuccessException.java
deleted file mode 100644
index 60e2e51..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/SuccessException.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.testutils;
-
-/**
- * Exception that is thrown to terminate a program and indicate success.
- */
-public class SuccessException extends Exception {
- private static final long serialVersionUID = -7011865671593955887L;
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ThrottledMapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ThrottledMapper.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ThrottledMapper.java
deleted file mode 100644
index 872d42f..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ThrottledMapper.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.testutils;
-
-import org.apache.flink.api.common.functions.MapFunction;
-
-/**
- * An identity map function that sleeps between elements, throttling the
- * processing speed.
- *
- * @param <T> The type mapped.
- */
-public class ThrottledMapper<T> implements MapFunction<T,T> {
-
- private static final long serialVersionUID = 467008933767159126L;
-
- private final int sleep;
-
- public ThrottledMapper(int sleep) {
- this.sleep = sleep;
- }
-
- @Override
- public T map(T value) throws Exception {
- Thread.sleep(this.sleep);
- return value;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/Tuple2Partitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/Tuple2Partitioner.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/Tuple2Partitioner.java
deleted file mode 100644
index 1e5f027..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/Tuple2Partitioner.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.testutils;
-
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.streaming.connectors.SerializableKafkaPartitioner;
-
-/**
- * Special partitioner that uses the first field of a 2-tuple as the partition,
- * and that expects a specific number of partitions.
- */
-public class Tuple2Partitioner implements SerializableKafkaPartitioner {
-
- private static final long serialVersionUID = 1L;
-
- private final int expectedPartitions;
-
-
- public Tuple2Partitioner(int expectedPartitions) {
- this.expectedPartitions = expectedPartitions;
- }
-
- @Override
- public int partition(Object key, int numPartitions) {
- if (numPartitions != expectedPartitions) {
- throw new IllegalArgumentException("Expected " + expectedPartitions + " partitions");
- }
- @SuppressWarnings("unchecked")
- Tuple2<Integer, Integer> element = (Tuple2<Integer, Integer>) key;
-
- return element.f0;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ValidatingExactlyOnceSink.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ValidatingExactlyOnceSink.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ValidatingExactlyOnceSink.java
deleted file mode 100644
index a3c35fc..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/java/org/apache/flink/streaming/connectors/testutils/ValidatingExactlyOnceSink.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.testutils;
-
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.streaming.api.checkpoint.Checkpointed;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-
-import java.util.BitSet;
-
-public class ValidatingExactlyOnceSink implements SinkFunction<Integer>, Checkpointed<Tuple2<Integer, BitSet>> {
-
- private static final long serialVersionUID = 1748426382527469932L;
-
- private final int numElementsTotal;
-
- private BitSet duplicateChecker = new BitSet(); // this is checkpointed
-
- private int numElements; // this is checkpointed
-
-
- public ValidatingExactlyOnceSink(int numElementsTotal) {
- this.numElementsTotal = numElementsTotal;
- }
-
-
- @Override
- public void invoke(Integer value) throws Exception {
- numElements++;
-
- if (duplicateChecker.get(value)) {
- throw new Exception("Received a duplicate");
- }
- duplicateChecker.set(value);
-
- if (numElements == numElementsTotal) {
- // validate
- if (duplicateChecker.cardinality() != numElementsTotal) {
- throw new Exception("Duplicate checker has wrong cardinality");
- }
- else if (duplicateChecker.nextClearBit(0) != numElementsTotal) {
- throw new Exception("Received sparse sequence");
- }
- else {
- throw new SuccessException();
- }
- }
- }
-
- @Override
- public Tuple2<Integer, BitSet> snapshotState(long checkpointId, long checkpointTimestamp) {
- return new Tuple2<>(numElements, duplicateChecker);
- }
-
- @Override
- public void restoreState(Tuple2<Integer, BitSet> state) {
- this.numElements = state.f0;
- this.duplicateChecker = state.f1;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/log4j-test.properties
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/log4j-test.properties b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/log4j-test.properties
deleted file mode 100644
index ded15e9..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/log4j-test.properties
+++ /dev/null
@@ -1,29 +0,0 @@
-################################################################################
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-################################################################################
-
-log4j.rootLogger=INFo, testlogger
-
-log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
-log4j.appender.testlogger.target = System.err
-log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
-log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
-
-# suppress the irrelevant (wrong) warnings from the netty channel handler
-log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, testlogger
-
-
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/logback-test.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/logback-test.xml b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/logback-test.xml
deleted file mode 100644
index 45b3b92..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka-083/src/test/resources/logback-test.xml
+++ /dev/null
@@ -1,30 +0,0 @@
-<!--
- ~ Licensed to the Apache Software Foundation (ASF) under one
- ~ or more contributor license agreements. See the NOTICE file
- ~ distributed with this work for additional information
- ~ regarding copyright ownership. The ASF licenses this file
- ~ to you under the Apache License, Version 2.0 (the
- ~ "License"); you may not use this file except in compliance
- ~ with the License. You may obtain a copy of the License at
- ~
- ~ http://www.apache.org/licenses/LICENSE-2.0
- ~
- ~ Unless required by applicable law or agreed to in writing, software
- ~ distributed under the License is distributed on an "AS IS" BASIS,
- ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- ~ See the License for the specific language governing permissions and
- ~ limitations under the License.
- -->
-
-<configuration>
- <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern>
- </encoder>
- </appender>
-
- <root level="WARN">
- <appender-ref ref="STDOUT"/>
- </root>
- <logger name="org.apache.flink.streaming" level="WARN"/>
-</configuration>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/pom.xml
----------------------------------------------------------------------
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
index c369c61..bbbd2be 100644
--- 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
@@ -96,6 +96,14 @@ under the License.
<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>
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer.java
new file mode 100644
index 0000000..85c4c17
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer.java
@@ -0,0 +1,681 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors;
+
+import kafka.cluster.Broker;
+import kafka.common.ErrorMapping;
+import kafka.javaapi.PartitionMetadata;
+import kafka.javaapi.TopicMetadata;
+import kafka.javaapi.TopicMetadataRequest;
+import kafka.javaapi.consumer.SimpleConsumer;
+import org.apache.commons.collections.map.LinkedMap;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.checkpoint.CheckpointNotifier;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedAsynchronously;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.streaming.connectors.internals.Fetcher;
+import org.apache.flink.streaming.connectors.internals.LegacyFetcher;
+import org.apache.flink.streaming.connectors.internals.OffsetHandler;
+import org.apache.flink.streaming.connectors.internals.ZookeeperOffsetHandler;
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+
+import org.apache.flink.util.NetUtils;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * The Flink Kafka Consumer is a streaming data source that pulls a parallel data stream from
+ * Apache Kafka. The consumer can run in multiple parallel instances, each of which will pull
+ * data from one or more Kafka partitions.
+ *
+ * <p>The Flink Kafka Consumer participates in checkpointing and guarantees that no data is lost
+ * during a failure, and that the computation processes elements "exactly once".
+ * (Note: These guarantees naturally assume that Kafka itself does not loose any data.)</p>
+ *
+ * <p>To support a variety of Kafka brokers, protocol versions, and offset committing approaches,
+ * the Flink Kafka Consumer can be parametrized with a <i>fetcher</i> and an <i>offset handler</i>.</p>
+ *
+ * <h1>Fetcher</h1>
+ *
+ * <p>The fetcher is responsible to pull data from Kafka. Because Kafka has undergone a change in
+ * protocols and APIs, there are currently two fetchers available:</p>
+ *
+ * <ul>
+ * <li>{@link FetcherType#NEW_HIGH_LEVEL}: A fetcher based on the new Kafka consumer API.
+ * This fetcher is generally more robust, but works only with later versions of
+ * Kafka (> 0.8.2).</li>
+ *
+ * <li>{@link FetcherType#LEGACY_LOW_LEVEL}: A fetcher based on the old low-level consumer API.
+ * This fetcher is works also with older versions of Kafka (0.8.1). The fetcher interprets
+ * the old Kafka consumer properties, like:
+ * <ul>
+ * <li>socket.timeout.ms</li>
+ * <li>socket.receive.buffer.bytes</li>
+ * <li>fetch.message.max.bytes</li>
+ * <li>auto.offset.reset with the values "latest", "earliest" (unlike 0.8.2 behavior)</li>
+ * <li>fetch.wait.max.ms</li>
+ * </ul>
+ * </li>
+ * </ul>
+ *
+ * <h1>Offset handler</h1>
+ *
+ * <p>Offsets whose records have been read and are checkpointed will be committed back to Kafka / ZooKeeper
+ * by the offset handler. In addition, the offset handler finds the point where the source initially
+ * starts reading from the stream, when the streaming job is started.</p>
+ *
+ * <p>Currently, the source offers two different offset handlers exist:</p>
+ * <ul>
+ * <li>{@link OffsetStore#KAFKA}: Use this offset handler when the Kafka brokers are managing the offsets,
+ * and hence offsets need to be committed the Kafka brokers, rather than to ZooKeeper.
+ * Note that this offset handler works only on new versions of Kafka (0.8.2.x +) and
+ * with the {@link FetcherType#NEW_HIGH_LEVEL} fetcher.</li>
+ *
+ * <li>{@link OffsetStore#FLINK_ZOOKEEPER}: Use this offset handler when the offsets are managed
+ * by ZooKeeper, as in older versions of Kafka (0.8.1.x)</li>
+ * </ul>
+ *
+ * <p>Please note that Flink snapshots the offsets internally as part of its distributed checkpoints. The offsets
+ * committed to Kafka / ZooKeeper are only to bring the outside view of progress in sync with Flink's view
+ * of the progress. That way, monitoring and other jobs can get a view of how far the Flink Kafka consumer
+ * has consumed a topic.</p>
+ *
+ * <p><b>NOTE:</b> The implementation currently accesses partition metadata when the consumer
+ * is constructed. That means that the client that submits the program needs to be able to
+ * reach the Kafka brokers or ZooKeeper.</p>
+ */
+public class FlinkKafkaConsumer<T> extends RichParallelSourceFunction<T>
+ implements CheckpointNotifier, CheckpointedAsynchronously<long[]>, ResultTypeQueryable<T> {
+
+ /**
+ * The offset store defines how acknowledged offsets are committed back to Kafka. Different
+ * options include letting Flink periodically commit to ZooKeeper, or letting Kafka manage the
+ * offsets (new Kafka versions only).
+ */
+ public enum OffsetStore {
+
+ /**
+ * Let Flink manage the offsets. Flink will periodically commit them to Zookeeper (usually after
+ * successful checkpoints), in the same structure as Kafka 0.8.2.x
+ *
+ * <p>Use this mode when using the source with Kafka 0.8.1.x brokers.</p>
+ */
+ FLINK_ZOOKEEPER,
+
+ /**
+ * Use the mechanisms in Kafka to commit offsets. Depending on the Kafka configuration, different
+ * mechanism will be used (broker coordinator, zookeeper)
+ */
+ KAFKA
+ }
+
+ /**
+ * The fetcher type defines which code paths to use to pull data from teh Kafka broker.
+ */
+ public enum FetcherType {
+
+ /**
+ * The legacy fetcher uses Kafka's old low-level consumer API.
+ *
+ * <p>Use this fetcher for Kafka 0.8.1 brokers.</p>
+ */
+ LEGACY_LOW_LEVEL,
+
+ /**
+ * This fetcher uses a backport of the new consumer API to pull data from the Kafka broker.
+ * It is the fetcher that will be maintained in the future, and it already
+ * handles certain failure cases with less overhead than the legacy fetcher.
+ *
+ * <p>This fetcher works only Kafka 0.8.2 and 0.8.3 (and future versions).</p>
+ */
+ NEW_HIGH_LEVEL
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static final long serialVersionUID = -6272159445203409112L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaConsumer.class);
+
+ /** Magic number to define an unset offset. Negative offsets are not used by Kafka (invalid),
+ * and we pick a number that is probably (hopefully) not used by Kafka as a magic number for anything else. */
+ public static final long OFFSET_NOT_SET = -915623761776L;
+
+ /** The maximum number of pending non-committed checkpoints to track, to avoid memory leaks */
+ public static final int MAX_NUM_PENDING_CHECKPOINTS = 100;
+
+ /** Configuration key for the number of retries for getting the partition info */
+ public static final String GET_PARTITIONS_RETRIES_KEY = "flink.get-partitions.retry";
+
+ /** Default number of retries for getting the partition info. One retry means going through the full list of brokers */
+ public static final int DEFAULT_GET_PARTITIONS_RETRIES = 3;
+
+
+
+ // ------ Configuration of the Consumer -------
+
+ /** The offset store where this consumer commits safe offsets */
+ private final OffsetStore offsetStore;
+
+ /** The type of fetcher to be used to pull data from Kafka */
+ private final FetcherType fetcherType;
+
+ /** name of the topic consumed by this source */
+ private final String topic;
+
+ /** The properties to parametrize the Kafka consumer and ZooKeeper client */
+ private final Properties props;
+
+ /** The ids of the partitions that are read by this consumer */
+ private final int[] partitions;
+
+ /** The schema to convert between Kafka#s byte messages, and Flink's objects */
+ private final DeserializationSchema<T> valueDeserializer;
+
+ // ------ Runtime State -------
+
+ /** Data for pending but uncommitted checkpoints */
+ private final LinkedMap pendingCheckpoints = new LinkedMap();
+
+ /** The fetcher used to pull data from the Kafka brokers */
+ private transient Fetcher fetcher;
+
+ /** The committer that persists the committed offsets */
+ private transient OffsetHandler offsetHandler;
+
+ /** The partitions actually handled by this consumer */
+ private transient List<TopicPartition> subscribedPartitions;
+
+ /** The offsets of the last returned elements */
+ private transient long[] lastOffsets;
+
+ /** The latest offsets that have been committed to Kafka or ZooKeeper. These are never
+ * newer then the last offsets (Flink's internal view is fresher) */
+ private transient long[] commitedOffsets;
+
+ /** The offsets to restore to, if the consumer restores state from a checkpoint */
+ private transient long[] restoreToOffset;
+
+ private volatile boolean running = true;
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a new Flink Kafka Consumer, using the given type of fetcher and offset handler.
+ *
+ * <p>To determine which kink of fetcher and offset handler to use, please refer to the docs
+ * at the beginnign of this class.</p>
+ *
+ * @param topic
+ * The Kafka topic to read from.
+ * @param valueDeserializer
+ * The deserializer to turn raw byte messages into Java/Scala objects.
+ * @param props
+ * The properties that are used to configure both the fetcher and the offset handler.
+ * @param offsetStore
+ * The type of offset store to use (Kafka / ZooKeeper)
+ * @param fetcherType
+ * The type of fetcher to use (new high-level API, old low-level API).
+ */
+ public FlinkKafkaConsumer(String topic, DeserializationSchema<T> valueDeserializer, Properties props,
+ OffsetStore offsetStore, FetcherType fetcherType) {
+ this.offsetStore = checkNotNull(offsetStore);
+ this.fetcherType = checkNotNull(fetcherType);
+
+ if(fetcherType == FetcherType.NEW_HIGH_LEVEL) {
+ throw new UnsupportedOperationException("The fetcher for Kafka 0.8.3 is not yet " +
+ "supported in Flink");
+ }
+ if (offsetStore == OffsetStore.KAFKA && fetcherType == FetcherType.LEGACY_LOW_LEVEL) {
+ throw new IllegalArgumentException(
+ "The Kafka offset handler cannot be used together with the old low-level fetcher.");
+ }
+
+ this.topic = checkNotNull(topic, "topic");
+ this.props = checkNotNull(props, "props");
+ this.valueDeserializer = checkNotNull(valueDeserializer, "valueDeserializer");
+
+ // validate the zookeeper properties
+ if (offsetStore == OffsetStore.FLINK_ZOOKEEPER) {
+ validateZooKeeperConfig(props);
+ }
+
+ // Connect to a broker to get the partitions
+ List<PartitionInfo> partitionInfos = getPartitionsForTopic(topic, props);
+
+ // get initial partitions list. The order of the partitions is important for consistent
+ // partition id assignment in restart cases.
+ this.partitions = new int[partitionInfos.size()];
+ for (int i = 0; i < partitionInfos.size(); i++) {
+ partitions[i] = partitionInfos.get(i).partition();
+
+ if (partitions[i] >= partitions.length) {
+ throw new RuntimeException("Kafka partition numbers are sparse");
+ }
+ }
+ LOG.info("Topic {} has {} partitions", topic, partitions.length);
+
+ // make sure that we take care of the committing
+ props.setProperty("enable.auto.commit", "false");
+ }
+
+ // ------------------------------------------------------------------------
+ // Source life cycle
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+
+ final int numConsumers = getRuntimeContext().getNumberOfParallelSubtasks();
+ final int thisComsumerIndex = getRuntimeContext().getIndexOfThisSubtask();
+
+ // pick which partitions we work on
+ subscribedPartitions = assignPartitions(this.partitions, this.topic, numConsumers, thisComsumerIndex);
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Kafka consumer {} will read partitions {} out of partitions {}",
+ thisComsumerIndex, subscribedPartitions, Arrays.toString(partitions));
+ }
+
+ // we leave the fetcher as null, if we have no partitions
+ if (subscribedPartitions.isEmpty()) {
+ LOG.info("Kafka consumer {} has no partitions (empty source)", thisComsumerIndex);
+ return;
+ }
+
+ // create fetcher
+ switch (fetcherType){
+ case NEW_HIGH_LEVEL:
+ throw new UnsupportedOperationException("Currently unsupported");
+ case LEGACY_LOW_LEVEL:
+ fetcher = new LegacyFetcher(topic, props, getRuntimeContext().getTaskName());
+ break;
+ default:
+ throw new RuntimeException("Requested unknown fetcher " + fetcher);
+ }
+ fetcher.setPartitionsToRead(subscribedPartitions);
+
+ // offset handling
+ switch (offsetStore){
+ case FLINK_ZOOKEEPER:
+ offsetHandler = new ZookeeperOffsetHandler(props);
+ break;
+ case KAFKA:
+ throw new Exception("Kafka offset handler cannot work with legacy fetcher");
+ default:
+ throw new RuntimeException("Requested unknown offset store " + offsetStore);
+ }
+
+ // set up operator state
+ lastOffsets = new long[partitions.length];
+ commitedOffsets = new long[partitions.length];
+
+ Arrays.fill(lastOffsets, OFFSET_NOT_SET);
+ Arrays.fill(commitedOffsets, OFFSET_NOT_SET);
+
+ // seek to last known pos, from restore request
+ if (restoreToOffset != null) {
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Consumer {} found offsets from previous checkpoint: {}",
+ thisComsumerIndex, Arrays.toString(restoreToOffset));
+ }
+
+ for (int i = 0; i < restoreToOffset.length; i++) {
+ long restoredOffset = restoreToOffset[i];
+ if (restoredOffset != OFFSET_NOT_SET) {
+ // if this fails because we are not subscribed to the topic, then the
+ // partition assignment is not deterministic!
+
+ // we set the offset +1 here, because seek() is accepting the next offset to read,
+ // but the restore offset is the last read offset
+ fetcher.seek(new TopicPartition(topic, i), restoredOffset + 1);
+ lastOffsets[i] = restoredOffset;
+ }
+ }
+ }
+ else {
+ // no restore request. Let the offset handler take care of the initial offset seeking
+ offsetHandler.seekFetcherToInitialOffsets(subscribedPartitions, fetcher);
+ }
+ }
+
+ @Override
+ public void run(SourceContext<T> sourceContext) throws Exception {
+ if (fetcher != null) {
+ fetcher.run(sourceContext, valueDeserializer, lastOffsets);
+ }
+ else {
+ // this source never completes
+ final Object waitLock = new Object();
+ while (running) {
+ // wait until we are canceled
+ try {
+ //noinspection SynchronizationOnLocalVariableOrMethodParameter
+ synchronized (waitLock) {
+ waitLock.wait();
+ }
+ }
+ catch (InterruptedException e) {
+ // do nothing, check our "running" status
+ }
+ }
+ }
+
+ // close the context after the work was done. this can actually only
+ // happen when the fetcher decides to stop fetching
+ sourceContext.close();
+ }
+
+ @Override
+ public void cancel() {
+ // set ourselves as not running
+ running = false;
+
+ // close the fetcher to interrupt any work
+ Fetcher fetcher = this.fetcher;
+ this.fetcher = null;
+ if (fetcher != null) {
+ try {
+ fetcher.close();
+ }
+ catch (IOException e) {
+ LOG.warn("Error while closing Kafka connector data fetcher", e);
+ }
+ }
+
+ OffsetHandler offsetHandler = this.offsetHandler;
+ this.offsetHandler = null;
+ if (offsetHandler != null) {
+ try {
+ offsetHandler.close();
+ }
+ catch (IOException e) {
+ LOG.warn("Error while closing Kafka connector offset handler", e);
+ }
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ cancel();
+ super.close();
+ }
+
+ @Override
+ public TypeInformation<T> getProducedType() {
+ return valueDeserializer.getProducedType();
+ }
+
+ // ------------------------------------------------------------------------
+ // Checkpoint and restore
+ // ------------------------------------------------------------------------
+
+ @Override
+ public long[] snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
+ if (lastOffsets == null) {
+ LOG.debug("snapshotState() requested on not yet opened source; returning null.");
+ return null;
+ }
+ if (!running) {
+ LOG.debug("snapshotState() called on closed source");
+ return null;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Snapshotting state. Offsets: {}, checkpoint id: {}, timestamp: {}",
+ Arrays.toString(lastOffsets), checkpointId, checkpointTimestamp);
+ }
+
+ long[] currentOffsets = Arrays.copyOf(lastOffsets, lastOffsets.length);
+
+ // the map cannot be asynchronously updated, because only one checkpoint call can happen
+ // on this function at a time: either snapshotState() or notifyCheckpointComplete()
+ pendingCheckpoints.put(checkpointId, currentOffsets);
+
+ while (pendingCheckpoints.size() > MAX_NUM_PENDING_CHECKPOINTS) {
+ pendingCheckpoints.remove(0);
+ }
+
+ return currentOffsets;
+ }
+
+ @Override
+ public void restoreState(long[] restoredOffsets) {
+ restoreToOffset = restoredOffsets;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ if (fetcher == null) {
+ LOG.debug("notifyCheckpointComplete() called on uninitialized source");
+ return;
+ }
+ if (!running) {
+ LOG.debug("notifyCheckpointComplete() called on closed source");
+ return;
+ }
+
+ // only one commit operation must be in progress
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Committing offsets externally for checkpoint {}", checkpointId);
+ }
+
+ long[] checkpointOffsets;
+
+ // the map may be asynchronously updates when snapshotting state, so we synchronize
+ synchronized (pendingCheckpoints) {
+ final int posInMap = pendingCheckpoints.indexOf(checkpointId);
+ if (posInMap == -1) {
+ LOG.warn("Received confirmation for unknown checkpoint id {}", checkpointId);
+ return;
+ }
+
+ checkpointOffsets = (long[]) pendingCheckpoints.remove(posInMap);
+
+ // remove older checkpoints in map
+ for (int i = 0; i < posInMap; i++) {
+ pendingCheckpoints.remove(0);
+ }
+ }
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Committing offsets {} to offset store: {}", Arrays.toString(checkpointOffsets), offsetStore);
+ }
+
+ // build the map of (topic,partition) -> committed offset
+ Map<TopicPartition, Long> offsetsToCommit = new HashMap<>();
+ for (TopicPartition tp : subscribedPartitions) {
+
+ int partition = tp.partition();
+ long offset = checkpointOffsets[partition];
+ long lastCommitted = commitedOffsets[partition];
+
+ if (offset != OFFSET_NOT_SET) {
+ if (offset > lastCommitted) {
+ offsetsToCommit.put(tp, offset);
+ LOG.debug("Committing offset {} for partition {}", offset, partition);
+ }
+ else {
+ LOG.debug("Ignoring offset {} for partition {} because it is already committed", offset, partition);
+ }
+ }
+ }
+
+ offsetHandler.commit(offsetsToCommit);
+ }
+
+ // ------------------------------------------------------------------------
+ // Miscellaneous utilities
+ // ------------------------------------------------------------------------
+
+ protected static List<TopicPartition> assignPartitions(int[] partitions, String topicName,
+ int numConsumers, int consumerIndex) {
+ checkArgument(numConsumers > 0);
+ checkArgument(consumerIndex < numConsumers);
+
+ List<TopicPartition> partitionsToSub = new ArrayList<>();
+
+ for (int i = 0; i < partitions.length; i++) {
+ if (i % numConsumers == consumerIndex) {
+ partitionsToSub.add(new TopicPartition(topicName, partitions[i]));
+ }
+ }
+ return partitionsToSub;
+ }
+
+ // ------------------------------------------------------------------------
+ // Kafka / ZooKeeper communication utilities
+ // ------------------------------------------------------------------------
+
+ /**
+ * Send request to Kafka to get partitions for topic.
+ *
+ * @param topic The name of the topic.
+ * @param properties The properties for the Kafka Consumer that is used to query the partitions for the topic.
+ */
+ public static List<PartitionInfo> getPartitionsForTopic(final String topic, final Properties properties) {
+ String seedBrokersConfString = properties.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG);
+ final int numRetries = Integer.valueOf(properties.getProperty(GET_PARTITIONS_RETRIES_KEY, Integer.toString(DEFAULT_GET_PARTITIONS_RETRIES)));
+
+ checkNotNull(seedBrokersConfString, "Configuration property " + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " not set");
+ String[] seedBrokers = seedBrokersConfString.split(",");
+ List<PartitionInfo> partitions = new ArrayList<>();
+
+ Random rnd = new Random();
+ retryLoop: for(int retry = 0; retry < numRetries; retry++) {
+ // we pick a seed broker randomly to avoid overloading the first broker with all the requests when the
+ // parallel source instances start. Still, we try all available brokers.
+ int index = rnd.nextInt(seedBrokers.length);
+ brokersLoop: for (int arrIdx = 0; arrIdx < seedBrokers.length; arrIdx++) {
+ String seedBroker = seedBrokers[index];
+ LOG.info("Trying to get topic metadata from broker {} in try {}/{}", seedBroker, retry, numRetries);
+ if (++index == seedBrokers.length) {
+ index = 0;
+ }
+
+ URL brokerUrl = NetUtils.getCorrectHostnamePort(seedBroker);
+ SimpleConsumer consumer = null;
+ try {
+ final String clientId = "flink-kafka-consumer-partition-lookup";
+ final int soTimeout = Integer.valueOf(properties.getProperty("socket.timeout.ms", "30000"));
+ final int bufferSize = Integer.valueOf(properties.getProperty("socket.receive.buffer.bytes", "65536"));
+ consumer = new SimpleConsumer(brokerUrl.getHost(), brokerUrl.getPort(), soTimeout, bufferSize, clientId);
+
+ List<String> topics = Collections.singletonList(topic);
+ TopicMetadataRequest req = new TopicMetadataRequest(topics);
+ kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);
+
+ List<TopicMetadata> metaData = resp.topicsMetadata();
+
+ // clear in case we have an incomplete list from previous tries
+ partitions.clear();
+ for (TopicMetadata item : metaData) {
+ if (item.errorCode() != ErrorMapping.NoError()) {
+ if (item.errorCode() == ErrorMapping.InvalidTopicCode() || item.errorCode() == ErrorMapping.UnknownTopicOrPartitionCode()) {
+ // fail hard if topic is unknown
+ throw new RuntimeException("Requested partitions for unknown topic", ErrorMapping.exceptionFor(item.errorCode()));
+ }
+ // warn and try more brokers
+ LOG.warn("Error while getting metadata from broker " + seedBroker + " to find partitions for " + topic,
+ ErrorMapping.exceptionFor(item.errorCode()));
+ continue brokersLoop;
+ }
+ if (!item.topic().equals(topic)) {
+ LOG.warn("Received metadata from topic " + item.topic() + " even though it was not requested. Skipping ...");
+ continue brokersLoop;
+ }
+ for (PartitionMetadata part : item.partitionsMetadata()) {
+ Node leader = brokerToNode(part.leader());
+ Node[] replicas = new Node[part.replicas().size()];
+ for (int i = 0; i < part.replicas().size(); i++) {
+ replicas[i] = brokerToNode(part.replicas().get(i));
+ }
+
+ Node[] ISRs = new Node[part.isr().size()];
+ for (int i = 0; i < part.isr().size(); i++) {
+ ISRs[i] = brokerToNode(part.isr().get(i));
+ }
+ PartitionInfo pInfo = new PartitionInfo(topic, part.partitionId(), leader, replicas, ISRs);
+ partitions.add(pInfo);
+ }
+ }
+ break retryLoop; // leave the loop through the brokers
+ } catch (Exception e) {
+ LOG.warn("Error communicating with broker " + seedBroker + " to find partitions for " + topic, e);
+ } finally {
+ if (consumer != null) {
+ consumer.close();
+ }
+ }
+ } // brokers loop
+ } // retries loop
+ return partitions;
+ }
+
+ private static Node brokerToNode(Broker broker) {
+ return new Node(broker.id(), broker.host(), broker.port());
+ }
+
+ protected static void validateZooKeeperConfig(Properties props) {
+ if (props.getProperty("zookeeper.connect") == null) {
+ throw new IllegalArgumentException("Required property 'zookeeper.connect' has not been set in the properties");
+ }
+ if (props.getProperty(ConsumerConfig.GROUP_ID_CONFIG) == null) {
+ throw new IllegalArgumentException("Required property '" + ConsumerConfig.GROUP_ID_CONFIG
+ + "' has not been set in the properties");
+ }
+
+ try {
+ //noinspection ResultOfMethodCallIgnored
+ Integer.parseInt(props.getProperty("zookeeper.session.timeout.ms", "0"));
+ }
+ catch (NumberFormatException e) {
+ throw new IllegalArgumentException("Property 'zookeeper.session.timeout.ms' is not a valid integer");
+ }
+
+ try {
+ //noinspection ResultOfMethodCallIgnored
+ Integer.parseInt(props.getProperty("zookeeper.connection.timeout.ms", "0"));
+ }
+ catch (NumberFormatException e) {
+ throw new IllegalArgumentException("Property 'zookeeper.connection.timeout.ms' is not a valid integer");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer081.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer081.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer081.java
new file mode 100644
index 0000000..f696893
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer081.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors;
+
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+
+import java.util.Properties;
+
+/**
+ * Creates a Kafka consumer compatible with reading from Kafka 0.8.1.x brokers.
+ * The consumer will internally use the old low-level Kafka API, and manually commit offsets
+ * partition offsets to ZooKeeper.
+ *
+ * <p>The following additional configuration values are available:</p>
+ * <ul>
+ * <li>socket.timeout.ms</li>
+ * <li>socket.receive.buffer.bytes</li>
+ * <li>fetch.message.max.bytes</li>
+ * <li>auto.offset.reset with the values "latest", "earliest" (unlike 0.8.2 behavior)</li>
+ * <li>fetch.wait.max.ms</li>
+ * </ul>
+ *
+ * @param <T> The type of elements produced by this consumer.
+ */
+public class FlinkKafkaConsumer081<T> extends FlinkKafkaConsumer<T> {
+
+ private static final long serialVersionUID = -5649906773771949146L;
+
+ /**
+ * Creates a new Kafka 0.8.1.x streaming source consumer.
+ *
+ * @param topic
+ * The name of the topic that should be consumed.
+ * @param valueDeserializer
+ * The de-/serializer used to convert between Kafka's byte messages and Flink's objects.
+ * @param props
+ * The properties used to configure the Kafka consumer client, and the ZooKeeper client.
+ */
+ public FlinkKafkaConsumer081(String topic, DeserializationSchema<T> valueDeserializer, Properties props) {
+ super(topic, valueDeserializer, props, OffsetStore.FLINK_ZOOKEEPER, FetcherType.LEGACY_LOW_LEVEL);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer082.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer082.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer082.java
new file mode 100644
index 0000000..ab1dc59
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/FlinkKafkaConsumer082.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors;
+
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+
+import java.util.Properties;
+
+/**
+ * Creates a Kafka consumer compatible with reading from Kafka 0.8.2.x brokers.
+ * The consumer will internally use the old low-level Kafka API, and manually commit offsets
+ * partition offsets to ZooKeeper.
+ *
+ * Once Kafka released the new consumer with Kafka 0.8.3 Flink might use the 0.8.3 consumer API
+ * also against Kafka 0.8.2 installations.
+ *
+ * @param <T> The type of elements produced by this consumer.
+ */
+public class FlinkKafkaConsumer082<T> extends FlinkKafkaConsumer<T> {
+
+ private static final long serialVersionUID = -8450689820627198228L;
+
+ /**
+ * Creates a new Kafka 0.8.2.x streaming source consumer.
+ *
+ * @param topic
+ * The name of the topic that should be consumed.
+ * @param valueDeserializer
+ * The de-/serializer used to convert between Kafka's byte messages and Flink's objects.
+ * @param props
+ * The properties used to configure the Kafka consumer client, and the ZooKeeper client.
+ */
+ public FlinkKafkaConsumer082(String topic, DeserializationSchema<T> valueDeserializer, Properties props) {
+ super(topic, valueDeserializer, props, OffsetStore.FLINK_ZOOKEEPER, FetcherType.LEGACY_LOW_LEVEL);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/KafkaSink.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/KafkaSink.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/KafkaSink.java
new file mode 100644
index 0000000..6d39541
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/KafkaSink.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors;
+
+import com.google.common.base.Preconditions;
+import kafka.javaapi.producer.Producer;
+import kafka.producer.KeyedMessage;
+import kafka.producer.ProducerConfig;
+import kafka.serializer.DefaultEncoder;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.apache.flink.util.NetUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Properties;
+
+
+/**
+ * Sink that emits its inputs to a Kafka topic.
+ *
+ * @param <IN>
+ * Type of the sink input
+ */
+public class KafkaSink<IN> extends RichSinkFunction<IN> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaSink.class);
+
+ private Producer<IN, byte[]> producer;
+ private Properties userDefinedProperties;
+ private String topicId;
+ private String brokerList;
+ private SerializationSchema<IN, byte[]> schema;
+ private SerializableKafkaPartitioner partitioner;
+ private Class<? extends SerializableKafkaPartitioner> partitionerClass = null;
+
+ /**
+ * Creates a KafkaSink for a given topic. The sink produces its input to
+ * the topic.
+ *
+ * @param brokerList
+ * Addresses of the brokers
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined serialization schema.
+ */
+ public KafkaSink(String brokerList, String topicId, SerializationSchema<IN, byte[]> serializationSchema) {
+ this(brokerList, topicId, new Properties(), serializationSchema);
+ }
+
+ /**
+ * Creates a KafkaSink for a given topic with custom Producer configuration.
+ * If you use this constructor, the broker should be set with the "metadata.broker.list"
+ * configuration.
+ *
+ * @param brokerList
+ * Addresses of the brokers
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param producerConfig
+ * Configurations of the Kafka producer
+ * @param serializationSchema
+ * User defined serialization schema.
+ */
+ public KafkaSink(String brokerList, String topicId, Properties producerConfig, SerializationSchema<IN, byte[]> serializationSchema) {
+ String[] elements = brokerList.split(",");
+ for(String broker: elements) {
+ NetUtils.getCorrectHostnamePort(broker);
+ }
+ Preconditions.checkNotNull(topicId, "TopicID not set");
+
+ this.brokerList = brokerList;
+ this.topicId = topicId;
+ this.schema = serializationSchema;
+ this.partitionerClass = null;
+ this.userDefinedProperties = producerConfig;
+ }
+
+ /**
+ * Creates a KafkaSink for a given topic. The sink produces its input to
+ * the topic.
+ *
+ * @param brokerList
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined serialization schema.
+ * @param partitioner
+ * User defined partitioner.
+ */
+ public KafkaSink(String brokerList, String topicId, SerializationSchema<IN, byte[]> serializationSchema, SerializableKafkaPartitioner partitioner) {
+ this(brokerList, topicId, serializationSchema);
+ ClosureCleaner.ensureSerializable(partitioner);
+ this.partitioner = partitioner;
+ }
+
+ public KafkaSink(String brokerList,
+ String topicId,
+ SerializationSchema<IN, byte[]> serializationSchema,
+ Class<? extends SerializableKafkaPartitioner> partitioner) {
+ this(brokerList, topicId, serializationSchema);
+ this.partitionerClass = partitioner;
+ }
+
+ /**
+ * Initializes the connection to Kafka.
+ */
+ @Override
+ public void open(Configuration configuration) {
+
+ Properties properties = new Properties();
+
+ properties.put("metadata.broker.list", brokerList);
+ properties.put("request.required.acks", "-1");
+ properties.put("message.send.max.retries", "10");
+
+ properties.put("serializer.class", DefaultEncoder.class.getCanonicalName());
+
+ // this will not be used as the key will not be serialized
+ properties.put("key.serializer.class", DefaultEncoder.class.getCanonicalName());
+
+ for (Map.Entry<Object, Object> propertiesEntry : userDefinedProperties.entrySet()) {
+ properties.put(propertiesEntry.getKey(), propertiesEntry.getValue());
+ }
+
+ if (partitioner != null) {
+ properties.put("partitioner.class", PartitionerWrapper.class.getCanonicalName());
+ // java serialization will do the rest.
+ properties.put(PartitionerWrapper.SERIALIZED_WRAPPER_NAME, partitioner);
+ }
+ if (partitionerClass != null) {
+ properties.put("partitioner.class", partitionerClass);
+ }
+
+ ProducerConfig config = new ProducerConfig(properties);
+
+ try {
+ producer = new Producer<IN, byte[]>(config);
+ } catch (NullPointerException e) {
+ throw new RuntimeException("Cannot connect to Kafka broker " + brokerList, e);
+ }
+ }
+
+ /**
+ * Called when new data arrives to the sink, and forwards it to Kafka.
+ *
+ * @param next
+ * The incoming data
+ */
+ @Override
+ public void invoke(IN next) {
+ byte[] serialized = schema.serialize(next);
+
+ // Sending message without serializable key.
+ producer.send(new KeyedMessage<IN, byte[]>(topicId, null, next, serialized));
+ }
+
+ @Override
+ public void close() {
+ if (producer != null) {
+ producer.close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/PartitionerWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/PartitionerWrapper.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/PartitionerWrapper.java
new file mode 100644
index 0000000..3a83e18
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/PartitionerWrapper.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors;
+
+import kafka.producer.Partitioner;
+import kafka.utils.VerifiableProperties;
+
+/**
+ * Hacky wrapper to send an object instance through a Properties - map.
+ *
+ * This works as follows:
+ * The recommended way of creating a KafkaSink is specifying a classname for the partitioner.
+ *
+ * Otherwise (if the user gave a (serializable) class instance), we give Kafka the PartitionerWrapper class of Flink.
+ * This is set in the key-value (java.util.Properties) map.
+ * In addition to that, we use the Properties.put(Object, Object) to store the instance of the (serializable).
+ * This is a hack because the put() method is called on the underlying Hashmap.
+ *
+ * This PartitionerWrapper is called with the Properties. From there, we extract the wrapped Partitioner instance.
+ *
+ * The serializable PartitionerWrapper is serialized into the Properties Hashmap and also deserialized from there.
+ */
+public class PartitionerWrapper implements Partitioner {
+ public final static String SERIALIZED_WRAPPER_NAME = "flink.kafka.wrapper.serialized";
+
+ private Partitioner wrapped;
+ public PartitionerWrapper(VerifiableProperties properties) {
+ wrapped = (Partitioner) properties.props().get(SERIALIZED_WRAPPER_NAME);
+ }
+
+ @Override
+ public int partition(Object value, int numberOfPartitions) {
+ return wrapped.partition(value, numberOfPartitions);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/SerializableKafkaPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/SerializableKafkaPartitioner.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/SerializableKafkaPartitioner.java
new file mode 100644
index 0000000..aff0c3a
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/SerializableKafkaPartitioner.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.connectors;
+
+import kafka.producer.Partitioner;
+
+import java.io.Serializable;
+
+public interface SerializableKafkaPartitioner extends Serializable, Partitioner {
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/76fcaca8/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/internals/Fetcher.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/internals/Fetcher.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/internals/Fetcher.java
new file mode 100644
index 0000000..38487a3
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/internals/Fetcher.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.internals;
+
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+import org.apache.kafka.common.TopicPartition;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * A fetcher pulls data from Kafka, from a fix set of partitions.
+ * The fetcher supports "seeking" inside the partitions, i.e., moving to a different offset.
+ */
+public interface Fetcher {
+
+ /**
+ * Set which partitions the fetcher should pull from.
+ *
+ * @param partitions The list of partitions for a topic that the fetcher will pull from.
+ */
+ void setPartitionsToRead(List<TopicPartition> partitions);
+
+ /**
+ * Closes the fetcher. This will stop any operation in the
+ * {@link #run(SourceFunction.SourceContext, DeserializationSchema, long[])} method and eventually
+ * close underlying connections and release all resources.
+ */
+ void close() throws IOException;
+
+ /**
+ * Starts fetch data from Kafka and emitting it into the stream.
+ *
+ * <p>To provide exactly once guarantees, the fetcher needs emit a record and update the update
+ * of the last consumed offset in one atomic operation:</p>
+ * <pre>{@code
+ *
+ * while (running) {
+ * T next = ...
+ * long offset = ...
+ * int partition = ...
+ * synchronized (sourceContext.getCheckpointLock()) {
+ * sourceContext.collect(next);
+ * lastOffsets[partition] = offset;
+ * }
+ * }
+ * }</pre>
+ *
+ * @param sourceContext The source context to emit elements to.
+ * @param valueDeserializer The deserializer to decode the raw values with.
+ * @param lastOffsets The array into which to store the offsets foe which elements are emitted.
+ *
+ * @param <T> The type of elements produced by the fetcher and emitted to the source context.
+ */
+ <T> void run(SourceFunction.SourceContext<T> sourceContext, DeserializationSchema<T> valueDeserializer,
+ long[] lastOffsets) throws Exception;
+
+ /**
+ * Set the next offset to read from for the given partition.
+ * For example, if the partition <i>i</i> offset is set to <i>n</i>, the Fetcher's next result
+ * will be the message with <i>offset=n</i>.
+ *
+ * @param topicPartition The partition for which to seek the offset.
+ * @param offsetToRead To offset to seek to.
+ */
+ void seek(TopicPartition topicPartition, long offsetToRead);
+}