You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2016/10/11 08:05:11 UTC
[1/3] flink git commit: [FLINK-4035] Refactor the Kafka 0.10
connector to be based upon the 0.9 connector
Repository: flink
Updated Branches:
refs/heads/master a079259f3 -> 6731ec1e4
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-streaming-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index 213ba4a..9d8fa9a 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -29,6 +29,8 @@ import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.io.FileUtils;
import org.apache.curator.test.TestingServer;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.connectors.kafka.testutils.ZooKeeperStringSerializer;
import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
@@ -94,10 +96,10 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> FlinkKafkaProducerBase<T> getProducer(String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) {
+ public <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) {
FlinkKafkaProducer09<T> prod = new FlinkKafkaProducer09<>(topic, serSchema, props, partitioner);
prod.setFlushOnCheckpoint(true);
- return prod;
+ return stream.addSink(prod);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
index 5b18c75..05028e6 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
@@ -21,6 +21,7 @@ import org.apache.commons.collections.map.LinkedMap;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.CheckpointListener;
@@ -36,6 +37,7 @@ import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionState;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.util.SerializedValue;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -169,6 +171,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
throw new IllegalStateException("A periodic watermark emitter has already been set.");
}
try {
+ ClosureCleaner.clean(assigner, true);
this.punctuatedWatermarkAssigner = new SerializedValue<>(assigner);
return this;
} catch (Exception e) {
@@ -203,6 +206,7 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
throw new IllegalStateException("A punctuated watermark emitter has already been set.");
}
try {
+ ClosureCleaner.clean(assigner, true);
this.periodicWatermarkAssigner = new SerializedValue<>(assigner);
return this;
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
index f0975dc..26a695e 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
@@ -105,7 +105,7 @@ public abstract class FlinkKafkaProducerBase<IN> extends RichSinkFunction<IN> im
/**
* If true, the producer will wait until all outstanding records have been send to the broker.
*/
- private boolean flushOnCheckpoint;
+ protected boolean flushOnCheckpoint;
// -------------------------------- Runtime fields ------------------------------------------
@@ -139,7 +139,7 @@ public abstract class FlinkKafkaProducerBase<IN> extends RichSinkFunction<IN> im
requireNonNull(defaultTopicId, "TopicID not set");
requireNonNull(serializationSchema, "serializationSchema not set");
requireNonNull(producerConfig, "producerConfig not set");
- ClosureCleaner.ensureSerializable(customPartitioner);
+ ClosureCleaner.clean(customPartitioner, true);
ClosureCleaner.ensureSerializable(serializationSchema);
this.defaultTopicId = defaultTopicId;
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
index 7ce3a9d..7ee3079 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java
@@ -49,24 +49,24 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
public abstract class AbstractFetcher<T, KPH> {
- private static final int NO_TIMESTAMPS_WATERMARKS = 0;
- private static final int PERIODIC_WATERMARKS = 1;
- private static final int PUNCTUATED_WATERMARKS = 2;
+ protected static final int NO_TIMESTAMPS_WATERMARKS = 0;
+ protected static final int PERIODIC_WATERMARKS = 1;
+ protected static final int PUNCTUATED_WATERMARKS = 2;
// ------------------------------------------------------------------------
/** The source context to emit records and watermarks to */
- private final SourceContext<T> sourceContext;
+ protected final SourceContext<T> sourceContext;
/** The lock that guarantees that record emission and state updates are atomic,
* from the view of taking a checkpoint */
- private final Object checkpointLock;
+ protected final Object checkpointLock;
/** All partitions (and their state) that this fetcher is subscribed to */
private final KafkaTopicPartitionState<KPH>[] allPartitions;
/** The mode describing whether the fetcher also generates timestamps and watermarks */
- private final int timestampWatermarkMode;
+ protected final int timestampWatermarkMode;
/** Flag whether to register metrics for the fetcher */
protected final boolean useMetrics;
@@ -207,30 +207,34 @@ public abstract class AbstractFetcher<T, KPH> {
// ------------------------------------------------------------------------
/**
- *
* <p>Implementation Note: This method is kept brief to be JIT inlining friendly.
* That makes the fast path efficient, the extended paths are called as separate methods.
- *
* @param record The record to emit
* @param partitionState The state of the Kafka partition from which the record was fetched
- * @param offset The offset from which the record was fetched
+ * @param offset The offset of the record
+ * @param timestamp The record's event-timestamp
*/
- protected final void emitRecord(T record, KafkaTopicPartitionState<KPH> partitionState, long offset) {
+ protected void emitRecord(T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long timestamp) throws Exception {
if (timestampWatermarkMode == NO_TIMESTAMPS_WATERMARKS) {
// fast path logic, in case there are no watermarks
// emit the record, using the checkpoint lock to guarantee
// atomicity of record emission and offset state update
synchronized (checkpointLock) {
- sourceContext.collect(record);
+ if(timestamp != Long.MIN_VALUE) {
+ // this case is true for Kafka 0.10
+ sourceContext.collectWithTimestamp(record, timestamp);
+ } else {
+ sourceContext.collect(record);
+ }
partitionState.setOffset(offset);
}
}
else if (timestampWatermarkMode == PERIODIC_WATERMARKS) {
- emitRecordWithTimestampAndPeriodicWatermark(record, partitionState, offset);
+ emitRecordWithTimestampAndPeriodicWatermark(record, partitionState, offset, timestamp);
}
else {
- emitRecordWithTimestampAndPunctuatedWatermark(record, partitionState, offset);
+ emitRecordWithTimestampAndPunctuatedWatermark(record, partitionState, offset, timestamp);
}
}
@@ -238,8 +242,8 @@ public abstract class AbstractFetcher<T, KPH> {
* Record emission, if a timestamp will be attached from an assigner that is
* also a periodic watermark generator.
*/
- private void emitRecordWithTimestampAndPeriodicWatermark(
- T record, KafkaTopicPartitionState<KPH> partitionState, long offset)
+ protected void emitRecordWithTimestampAndPeriodicWatermark(
+ T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long kafkaEventTimestamp)
{
@SuppressWarnings("unchecked")
final KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH> withWatermarksState =
@@ -251,7 +255,7 @@ public abstract class AbstractFetcher<T, KPH> {
final long timestamp;
//noinspection SynchronizationOnLocalVariableOrMethodParameter
synchronized (withWatermarksState) {
- timestamp = withWatermarksState.getTimestampForRecord(record);
+ timestamp = withWatermarksState.getTimestampForRecord(record, kafkaEventTimestamp);
}
// emit the record with timestamp, using the usual checkpoint lock to guarantee
@@ -266,8 +270,8 @@ public abstract class AbstractFetcher<T, KPH> {
* Record emission, if a timestamp will be attached from an assigner that is
* also a punctuated watermark generator.
*/
- private void emitRecordWithTimestampAndPunctuatedWatermark(
- T record, KafkaTopicPartitionState<KPH> partitionState, long offset)
+ protected void emitRecordWithTimestampAndPunctuatedWatermark(
+ T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long kafkaEventTimestamp)
{
@SuppressWarnings("unchecked")
final KafkaTopicPartitionStateWithPunctuatedWatermarks<T, KPH> withWatermarksState =
@@ -275,7 +279,7 @@ public abstract class AbstractFetcher<T, KPH> {
// only one thread ever works on accessing timestamps and watermarks
// from the punctuated extractor
- final long timestamp = withWatermarksState.getTimestampForRecord(record);
+ final long timestamp = withWatermarksState.getTimestampForRecord(record, kafkaEventTimestamp);
final Watermark newWatermark = withWatermarksState.checkAndGetNewWatermark(record, timestamp);
// emit the record with timestamp, using the usual checkpoint lock to guarantee
@@ -291,6 +295,7 @@ public abstract class AbstractFetcher<T, KPH> {
updateMinPunctuatedWatermark(newWatermark);
}
}
+
/**
*Checks whether a new per-partition watermark is also a new cross-partition watermark.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPeriodicWatermarks.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPeriodicWatermarks.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPeriodicWatermarks.java
index 99c5d69..efdc73f 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPeriodicWatermarks.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPeriodicWatermarks.java
@@ -49,8 +49,8 @@ public final class KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH> extend
// ------------------------------------------------------------------------
- public long getTimestampForRecord (T record) {
- return timestampsAndWatermarks.extractTimestamp(record, Long.MIN_VALUE);
+ public long getTimestampForRecord(T record, long kafkaEventTimestamp) {
+ return timestampsAndWatermarks.extractTimestamp(record, kafkaEventTimestamp);
}
public long getCurrentWatermarkTimestamp() {
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPunctuatedWatermarks.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPunctuatedWatermarks.java b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPunctuatedWatermarks.java
index b265990..edf40ce 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPunctuatedWatermarks.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPunctuatedWatermarks.java
@@ -54,8 +54,8 @@ public final class KafkaTopicPartitionStateWithPunctuatedWatermarks<T, KPH> exte
// ------------------------------------------------------------------------
- public long getTimestampForRecord(T record) {
- return timestampsAndWatermarks.extractTimestamp(record, Long.MIN_VALUE);
+ public long getTimestampForRecord(T record, long kafkaEventTimestamp) {
+ return timestampsAndWatermarks.extractTimestamp(record, kafkaEventTimestamp);
}
@Nullable
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
index bafff4f..3c967ba 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
@@ -68,6 +68,7 @@ import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.functions.source.StatefulSequenceSource;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -91,6 +92,7 @@ import org.apache.flink.test.util.SuccessException;
import org.apache.flink.testutils.junit.RetryOnException;
import org.apache.flink.testutils.junit.RetryRule;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.StringUtils;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.errors.TimeoutException;
import org.junit.Assert;
@@ -185,7 +187,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
stream.print();
see.execute("No broker test");
} catch(ProgramInvocationException pie) {
- if(kafkaServer.getVersion().equals("0.9")) {
+ if(kafkaServer.getVersion().equals("0.9") || kafkaServer.getVersion().equals("0.10")) {
assertTrue(pie.getCause() instanceof JobExecutionException);
JobExecutionException jee = (JobExecutionException) pie.getCause();
@@ -287,8 +289,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
producerProperties.putAll(secureProps);
- FlinkKafkaProducerBase<Tuple2<Long, String>> prod = kafkaServer.getProducer(topic, new KeyedSerializationSchemaWrapper<>(sinkSchema), producerProperties, null);
- stream.addSink(prod);
+ kafkaServer.produceIntoKafka(stream, topic, new KeyedSerializationSchemaWrapper<>(sinkSchema), producerProperties, null);
// ----------- add consumer dataflow ----------
@@ -516,7 +517,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
// launch a producer thread
DataGenerators.InfiniteStringsGenerator generator =
- new DataGenerators.InfiniteStringsGenerator(kafkaServer, topic);
+ new DataGenerators.InfiniteStringsGenerator(kafkaServer, topic, flinkPort);
generator.start();
// launch a consumer asynchronously
@@ -539,7 +540,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
env.addSource(source).addSink(new DiscardingSink<String>());
- env.execute();
+ env.execute("Runner for CancelingOnFullInputTest");
}
catch (Throwable t) {
jobError.set(t);
@@ -560,7 +561,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
}
// cancel
- JobManagerCommunicationUtils.cancelCurrentJob(flink.getLeaderGateway(timeout));
+ JobManagerCommunicationUtils.cancelCurrentJob(flink.getLeaderGateway(timeout), "Runner for CancelingOnFullInputTest");
// wait for the program to be done and validate that we failed with the right exception
runnerThread.join();
@@ -570,6 +571,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
assertTrue(failueCause.getMessage().contains("Job was cancelled"));
if (generator.isAlive()) {
+ JobManagerCommunicationUtils.cancelCurrentJob(flink.getLeaderGateway(timeout), "String generator");
generator.shutdown();
generator.join();
}
@@ -613,7 +615,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
env.addSource(source).addSink(new DiscardingSink<String>());
- env.execute();
+ env.execute("CancelingOnEmptyInputTest");
}
catch (Throwable t) {
LOG.error("Job Runner failed with exception", t);
@@ -671,7 +673,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
.addSink(new DiscardingSink<Integer>());
try {
- env.execute();
+ env.execute("test fail on deploy");
fail("this test should fail with an exception");
}
catch (ProgramInvocationException e) {
@@ -738,8 +740,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
-
- stream.addSink(kafkaServer.getProducer("dummy", schema, props, null));
+ kafkaServer.produceIntoKafka(stream, "dummy", schema, props, null);
env.execute("Write to topics");
@@ -954,7 +955,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
}
});
- stream.addSink(kafkaServer.getProducer(topic, new KeyedSerializationSchemaWrapper<>(serSchema), producerProps, null));
+ kafkaServer.produceIntoKafka(stream, topic, new KeyedSerializationSchemaWrapper<>(serSchema), producerProps, null);
tryExecute(env, "big topology test");
deleteTestTopic(topic);
@@ -1046,8 +1047,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
KeyedSerializationSchema<Tuple2<Long, PojoValue>> schema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig());
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
- producerProperties.putAll(secureProps);
- kvStream.addSink(kafkaServer.getProducer(topic, schema, producerProperties, null));
+ kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write KV to Kafka");
// ----------- Read the data again -------------------
@@ -1132,8 +1132,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
producerProperties.putAll(secureProps);
-
- kvStream.addSink(kafkaServer.getProducer(topic, schema, producerProperties, null));
+ kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write deletes to Kafka");
@@ -1229,7 +1228,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
props.putAll(secureProps);
TypeInformationSerializationSchema<Tuple2<Integer, Integer>> schema = new TypeInformationSerializationSchema<>(TypeInfoParser.<Tuple2<Integer, Integer>>parse("Tuple2<Integer, Integer>"), env1.getConfig());
- DataStream<Tuple2<Integer, Integer>> fromKafka = env1.addSource(kafkaServer.getConsumer(topic, schema, props));
+ DataStream<Tuple2<Integer, Integer>> fromKafka = env1.addSource(kafkaServer.getConsumer(topic, schema, standardProps));
fromKafka.flatMap(new FlatMapFunction<Tuple2<Integer, Integer>, Void>() {
@Override
public void flatMap(Tuple2<Integer, Integer> value, Collector<Void> out) throws Exception {// no op
@@ -1254,7 +1253,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
}
});
- fromGen.addSink(kafkaServer.getProducer(topic, new KeyedSerializationSchemaWrapper<>(schema), props, null));
+ kafkaServer.produceIntoKafka(fromGen, topic, new KeyedSerializationSchemaWrapper<>(schema), standardProps, null);
env1.execute("Metrics test job");
} catch(Throwable t) {
@@ -1369,8 +1368,8 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
final TypeInformationSerializationSchema<Tuple2<Integer, Integer>> deser =
new TypeInformationSerializationSchema<>(intIntTupleType, env.getConfig());
- cc.putAll(secureProps);
// create the consumer
+ cc.putAll(secureProps);
FlinkKafkaConsumerBase<Tuple2<Integer, Integer>> consumer = kafkaServer.getConsumer(topicName, deser, cc);
DataStream<Tuple2<Integer, Integer>> source = env
@@ -1474,9 +1473,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
producerProperties.setProperty("retries", "0");
producerProperties.putAll(secureProps);
- stream.addSink(kafkaServer.getProducer(
- topicName, serSchema, producerProperties,
- new Tuple2Partitioner(parallelism)))
+ kafkaServer.produceIntoKafka(stream, topicName, serSchema, producerProperties, new Tuple2Partitioner(parallelism))
.setParallelism(parallelism);
try {
@@ -1773,86 +1770,88 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
// deleteTestTopic(topic);
// }
// }
-
- private void executeAndCatchException(StreamExecutionEnvironment env, String execName) throws Exception {
- try {
- tryExecutePropagateExceptions(env, execName);
- }
- catch (ProgramInvocationException | JobExecutionException e) {
- // look for NotLeaderForPartitionException
- Throwable cause = e.getCause();
-
- // search for nested SuccessExceptions
- int depth = 0;
- while (cause != null && depth++ < 20) {
- if (cause instanceof kafka.common.NotLeaderForPartitionException) {
- throw (Exception) cause;
- }
- cause = cause.getCause();
- }
- throw e;
- }
- }
-
- private void putDataInTopics(StreamExecutionEnvironment env,
- Properties producerProperties,
- final int elementsPerPartition,
- Map<String, Boolean> topics,
- TypeInformation<Tuple2<Long, Integer>> outputTypeInfo) {
- if(topics.size() != 2) {
- throw new RuntimeException("This method accepts two topics as arguments.");
- }
-
- TypeInformationSerializationSchema<Tuple2<Long, Integer>> sinkSchema =
- new TypeInformationSerializationSchema<>(outputTypeInfo, env.getConfig());
-
- DataStream<Tuple2<Long, Integer>> stream = env
- .addSource(new RichParallelSourceFunction<Tuple2<Long, Integer>>() {
- private boolean running = true;
-
- @Override
- public void run(SourceContext<Tuple2<Long, Integer>> ctx) throws InterruptedException {
- int topic = 0;
- int currentTs = 1;
-
- while (running && currentTs < elementsPerPartition) {
- long timestamp = (currentTs % 10 == 0) ? -1L : currentTs;
- ctx.collect(new Tuple2<Long, Integer>(timestamp, topic));
- currentTs++;
- }
-
- Tuple2<Long, Integer> toWrite2 = new Tuple2<Long, Integer>(-1L, topic);
- ctx.collect(toWrite2);
- }
-
- @Override
- public void cancel() {
- running = false;
- }
- }).setParallelism(1);
-
- List<Map.Entry<String, Boolean>> topicsL = new ArrayList<>(topics.entrySet());
- stream.map(new MapFunction<Tuple2<Long,Integer>, Tuple2<Long,Integer>>() {
-
- @Override
- public Tuple2<Long, Integer> map(Tuple2<Long, Integer> value) throws Exception {
- return value;
- }
- }).setParallelism(1).addSink(kafkaServer.getProducer(topicsL.get(0).getKey(),
- new KeyedSerializationSchemaWrapper<>(sinkSchema), producerProperties, null)).setParallelism(1);
-
- if(!topicsL.get(1).getValue()) {
- stream.map(new MapFunction<Tuple2<Long,Integer>, Tuple2<Long,Integer>>() {
-
- @Override
- public Tuple2<Long, Integer> map(Tuple2<Long, Integer> value) throws Exception {
- long timestamp = (value.f0 == -1) ? -1L : 1000 + value.f0;
- return new Tuple2<Long, Integer>(timestamp, 1);
- }
- }).setParallelism(1).addSink(kafkaServer.getProducer(topicsL.get(1).getKey(),
- new KeyedSerializationSchemaWrapper<>(sinkSchema), producerProperties, null)).setParallelism(1);
- }
- }
+//
+// private void executeAndCatchException(StreamExecutionEnvironment env, String execName) throws Exception {
+// try {
+// tryExecutePropagateExceptions(env, execName);
+// }
+// catch (ProgramInvocationException | JobExecutionException e) {
+// // look for NotLeaderForPartitionException
+// Throwable cause = e.getCause();
+//
+// // search for nested SuccessExceptions
+// int depth = 0;
+// while (cause != null && depth++ < 20) {
+// if (cause instanceof kafka.common.NotLeaderForPartitionException) {
+// throw (Exception) cause;
+// }
+// cause = cause.getCause();
+// }
+// throw e;
+// }
+// }
+//
+// private void putDataInTopics(StreamExecutionEnvironment env,
+// Properties producerProperties,
+// final int elementsPerPartition,
+// Map<String, Boolean> topics,
+// TypeInformation<Tuple2<Long, Integer>> outputTypeInfo) {
+// if(topics.size() != 2) {
+// throw new RuntimeException("This method accepts two topics as arguments.");
+// }
+//
+// TypeInformationSerializationSchema<Tuple2<Long, Integer>> sinkSchema =
+// new TypeInformationSerializationSchema<>(outputTypeInfo, env.getConfig());
+//
+// DataStream<Tuple2<Long, Integer>> stream = env
+// .addSource(new RichParallelSourceFunction<Tuple2<Long, Integer>>() {
+// private boolean running = true;
+//
+// @Override
+// public void run(SourceContext<Tuple2<Long, Integer>> ctx) throws InterruptedException {
+// int topic = 0;
+// int currentTs = 1;
+//
+// while (running && currentTs < elementsPerPartition) {
+// long timestamp = (currentTs % 10 == 0) ? -1L : currentTs;
+// ctx.collect(new Tuple2<Long, Integer>(timestamp, topic));
+// currentTs++;
+// }
+//
+// Tuple2<Long, Integer> toWrite2 = new Tuple2<Long, Integer>(-1L, topic);
+// ctx.collect(toWrite2);
+// }
+//
+// @Override
+// public void cancel() {
+// running = false;
+// }
+// }).setParallelism(1);
+//
+// List<Map.Entry<String, Boolean>> topicsL = new ArrayList<>(topics.entrySet());
+//
+// stream = stream.map(new MapFunction<Tuple2<Long,Integer>, Tuple2<Long,Integer>>() {
+//
+// @Override
+// public Tuple2<Long, Integer> map(Tuple2<Long, Integer> value) throws Exception {
+// return value;
+// }
+// }).setParallelism(1);
+// kafkaServer.produceIntoKafka(stream, topicsL.get(0).getKey(),
+// new KeyedSerializationSchemaWrapper<>(sinkSchema), producerProperties, null).setParallelism(1);
+//
+// if(!topicsL.get(1).getValue()) {
+// stream.map(new MapFunction<Tuple2<Long,Integer>, Tuple2<Long,Integer>>() {
+//
+// @Override
+// public Tuple2<Long, Integer> map(Tuple2<Long, Integer> value) throws Exception {
+// long timestamp = (value.f0 == -1) ? -1L : 1000 + value.f0;
+// return new Tuple2<>(timestamp, 1);
+// }
+// }).setParallelism(1).addSink(kafkaServer.produceIntoKafka(topicsL.get(1).getKey(),
+// new KeyedSerializationSchemaWrapper<>(sinkSchema), producerProperties, null)).setParallelism(1);
+// }
+// }
private DataStreamSink<Tuple2<Long, Integer>> runPunctuatedComsumer(StreamExecutionEnvironment env,
List<String> topics,
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
index 5bcf406..c925c8f 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
@@ -109,11 +109,10 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
props.putAll(secureProps);
// sink partitions into
- stream.addSink(kafkaServer.getProducer(topic,
+ kafkaServer.produceIntoKafka(stream, topic,
new KeyedSerializationSchemaWrapper<>(serSchema),
props,
- new CustomPartitioner(parallelism)))
- .setParallelism(parallelism);
+ new CustomPartitioner(parallelism)).setParallelism(parallelism);
// ------ consuming topology ---------
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java
index 9236e78..5c03b78 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java
@@ -165,12 +165,10 @@ public class KafkaShortRetentionTestBase implements Serializable {
running = false;
}
});
-
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
-
- stream.addSink(kafkaServer.getProducer(topic, new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), props, null));
+ kafkaServer.produceIntoKafka(stream, topic, new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), props, null);
// ----------- add consumer dataflow ----------
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
index 6ecde71..ded1fde 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
@@ -18,6 +18,9 @@
package org.apache.flink.streaming.connectors.kafka;
import kafka.server.KafkaServer;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
@@ -76,7 +79,9 @@ public abstract class KafkaTestEnvironment {
public abstract <T> FlinkKafkaConsumerBase<T> getConsumer(List<String> topics, KeyedDeserializationSchema<T> readSchema, Properties props);
- public abstract <T> FlinkKafkaProducerBase<T> getProducer(String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner);
+ public abstract <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic,
+ KeyedSerializationSchema<T> serSchema, Properties props,
+ KafkaPartitioner<T> partitioner);
// -- leader failure simulation
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTimestampsTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTimestampsTest.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTimestampsTest.java
index 9b5d2e6..582311f 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTimestampsTest.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcherTimestampsTest.java
@@ -46,10 +46,11 @@ public class AbstractFetcherTimestampsTest {
@Test
public void testPunctuatedWatermarks() throws Exception {
+ final String testTopic = "test topic name";
List<KafkaTopicPartition> originalPartitions = Arrays.asList(
- new KafkaTopicPartition("test topic name", 7),
- new KafkaTopicPartition("test topic name", 13),
- new KafkaTopicPartition("test topic name", 21));
+ new KafkaTopicPartition(testTopic, 7),
+ new KafkaTopicPartition(testTopic, 13),
+ new KafkaTopicPartition(testTopic, 21));
TestSourceContext<Long> sourceContext = new TestSourceContext<>();
@@ -65,22 +66,22 @@ public class AbstractFetcherTimestampsTest {
// elements generate a watermark if the timestamp is a multiple of three
// elements for partition 1
- fetcher.emitRecord(1L, part1, 1L);
- fetcher.emitRecord(2L, part1, 2L);
- fetcher.emitRecord(3L, part1, 3L);
+ fetcher.emitRecord(1L, part1, 1L, Long.MIN_VALUE);
+ fetcher.emitRecord(2L, part1, 2L, Long.MIN_VALUE);
+ fetcher.emitRecord(3L, part1, 3L, Long.MIN_VALUE);
assertEquals(3L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(3L, sourceContext.getLatestElement().getTimestamp());
assertFalse(sourceContext.hasWatermark());
// elements for partition 2
- fetcher.emitRecord(12L, part2, 1L);
+ fetcher.emitRecord(12L, part2, 1L, Long.MIN_VALUE);
assertEquals(12L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(12L, sourceContext.getLatestElement().getTimestamp());
assertFalse(sourceContext.hasWatermark());
// elements for partition 3
- fetcher.emitRecord(101L, part3, 1L);
- fetcher.emitRecord(102L, part3, 2L);
+ fetcher.emitRecord(101L, part3, 1L, Long.MIN_VALUE);
+ fetcher.emitRecord(102L, part3, 2L, Long.MIN_VALUE);
assertEquals(102L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(102L, sourceContext.getLatestElement().getTimestamp());
@@ -89,25 +90,25 @@ public class AbstractFetcherTimestampsTest {
assertEquals(3L, sourceContext.getLatestWatermark().getTimestamp());
// advance partition 3
- fetcher.emitRecord(1003L, part3, 3L);
- fetcher.emitRecord(1004L, part3, 4L);
- fetcher.emitRecord(1005L, part3, 5L);
+ fetcher.emitRecord(1003L, part3, 3L, Long.MIN_VALUE);
+ fetcher.emitRecord(1004L, part3, 4L, Long.MIN_VALUE);
+ fetcher.emitRecord(1005L, part3, 5L, Long.MIN_VALUE);
assertEquals(1005L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(1005L, sourceContext.getLatestElement().getTimestamp());
// advance partition 1 beyond partition 2 - this bumps the watermark
- fetcher.emitRecord(30L, part1, 4L);
+ fetcher.emitRecord(30L, part1, 4L, Long.MIN_VALUE);
assertEquals(30L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(30L, sourceContext.getLatestElement().getTimestamp());
assertTrue(sourceContext.hasWatermark());
assertEquals(12L, sourceContext.getLatestWatermark().getTimestamp());
// advance partition 2 again - this bumps the watermark
- fetcher.emitRecord(13L, part2, 2L);
+ fetcher.emitRecord(13L, part2, 2L, Long.MIN_VALUE);
assertFalse(sourceContext.hasWatermark());
- fetcher.emitRecord(14L, part2, 3L);
+ fetcher.emitRecord(14L, part2, 3L, Long.MIN_VALUE);
assertFalse(sourceContext.hasWatermark());
- fetcher.emitRecord(15L, part2, 3L);
+ fetcher.emitRecord(15L, part2, 3L, Long.MIN_VALUE);
assertTrue(sourceContext.hasWatermark());
assertEquals(15L, sourceContext.getLatestWatermark().getTimestamp());
}
@@ -117,11 +118,12 @@ public class AbstractFetcherTimestampsTest {
ExecutionConfig config = new ExecutionConfig();
config.setAutoWatermarkInterval(10);
-
+
+ final String testTopic = "test topic name";
List<KafkaTopicPartition> originalPartitions = Arrays.asList(
- new KafkaTopicPartition("test topic name", 7),
- new KafkaTopicPartition("test topic name", 13),
- new KafkaTopicPartition("test topic name", 21));
+ new KafkaTopicPartition(testTopic, 7),
+ new KafkaTopicPartition(testTopic, 13),
+ new KafkaTopicPartition(testTopic, 21));
TestSourceContext<Long> sourceContext = new TestSourceContext<>();
@@ -142,20 +144,20 @@ public class AbstractFetcherTimestampsTest {
// elements generate a watermark if the timestamp is a multiple of three
// elements for partition 1
- fetcher.emitRecord(1L, part1, 1L);
- fetcher.emitRecord(2L, part1, 2L);
- fetcher.emitRecord(3L, part1, 3L);
+ fetcher.emitRecord(1L, part1, 1L, Long.MIN_VALUE);
+ fetcher.emitRecord(2L, part1, 2L, Long.MIN_VALUE);
+ fetcher.emitRecord(3L, part1, 3L, Long.MIN_VALUE);
assertEquals(3L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(3L, sourceContext.getLatestElement().getTimestamp());
// elements for partition 2
- fetcher.emitRecord(12L, part2, 1L);
+ fetcher.emitRecord(12L, part2, 1L, Long.MIN_VALUE);
assertEquals(12L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(12L, sourceContext.getLatestElement().getTimestamp());
// elements for partition 3
- fetcher.emitRecord(101L, part3, 1L);
- fetcher.emitRecord(102L, part3, 2L);
+ fetcher.emitRecord(101L, part3, 1L, Long.MIN_VALUE);
+ fetcher.emitRecord(102L, part3, 2L, Long.MIN_VALUE);
assertEquals(102L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(102L, sourceContext.getLatestElement().getTimestamp());
@@ -163,14 +165,14 @@ public class AbstractFetcherTimestampsTest {
assertEquals(3L, sourceContext.getLatestWatermark().getTimestamp());
// advance partition 3
- fetcher.emitRecord(1003L, part3, 3L);
- fetcher.emitRecord(1004L, part3, 4L);
- fetcher.emitRecord(1005L, part3, 5L);
+ fetcher.emitRecord(1003L, part3, 3L, Long.MIN_VALUE);
+ fetcher.emitRecord(1004L, part3, 4L, Long.MIN_VALUE);
+ fetcher.emitRecord(1005L, part3, 5L, Long.MIN_VALUE);
assertEquals(1005L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(1005L, sourceContext.getLatestElement().getTimestamp());
// advance partition 1 beyond partition 2 - this bumps the watermark
- fetcher.emitRecord(30L, part1, 4L);
+ fetcher.emitRecord(30L, part1, 4L, Long.MIN_VALUE);
assertEquals(30L, sourceContext.getLatestElement().getValue().longValue());
assertEquals(30L, sourceContext.getLatestElement().getTimestamp());
@@ -178,9 +180,9 @@ public class AbstractFetcherTimestampsTest {
assertEquals(12L, sourceContext.getLatestWatermark().getTimestamp());
// advance partition 2 again - this bumps the watermark
- fetcher.emitRecord(13L, part2, 2L);
- fetcher.emitRecord(14L, part2, 3L);
- fetcher.emitRecord(15L, part2, 3L);
+ fetcher.emitRecord(13L, part2, 2L, Long.MIN_VALUE);
+ fetcher.emitRecord(14L, part2, 3L, Long.MIN_VALUE);
+ fetcher.emitRecord(15L, part2, 3L, Long.MIN_VALUE);
// this blocks until the periodic thread emitted the watermark
long watermarkTs = sourceContext.getLatestWatermark().getTimestamp();
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
index 58a5cc3..3f035fd 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
@@ -23,10 +23,10 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase;
import org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironment;
import org.apache.flink.streaming.connectors.kafka.partitioner.FixedPartitioner;
@@ -35,6 +35,7 @@ import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWra
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.streaming.util.serialization.TypeInformationSerializationSchema;
+import java.io.Serializable;
import java.util.Properties;
import java.util.Random;
@@ -63,7 +64,7 @@ public class DataGenerators {
int partition = getRuntimeContext().getIndexOfThisSubtask();
while (running && cnt <= to) {
- ctx.collect(new Tuple2<Integer, Integer>(partition, cnt));
+ ctx.collect(new Tuple2<>(partition, cnt));
cnt++;
}
}
@@ -74,11 +75,11 @@ public class DataGenerators {
}
});
- stream.addSink(testServer.getProducer(topic,
+ testServer.produceIntoKafka(stream, topic,
new KeyedSerializationSchemaWrapper<>(new TypeInformationSerializationSchema<>(resultType, env.getConfig())),
FlinkKafkaProducerBase.getPropertiesFromBrokerList(testServer.getBrokerConnectionString()),
new Tuple2Partitioner(numPartitions)
- ));
+ );
env.execute("Data generator (Int, Int) stream to topic " + topic);
}
@@ -141,38 +142,40 @@ public class DataGenerators {
if(secureProps != null) {
props.putAll(testServer.getSecureProperties());
}
-
- stream
- .rebalance()
- .addSink(testServer.getProducer(topic,
- new KeyedSerializationSchemaWrapper<>(new TypeInformationSerializationSchema<>(BasicTypeInfo.INT_TYPE_INFO, env.getConfig())),
- props,
- new KafkaPartitioner<Integer>() {
- @Override
- public int partition(Integer next, byte[] serializedKey, byte[] serializedValue, int numPartitions) {
- return next % numPartitions;
- }
- }));
+
+ stream = stream.rebalance();
+ testServer.produceIntoKafka(stream, topic,
+ new KeyedSerializationSchemaWrapper<>(new TypeInformationSerializationSchema<>(BasicTypeInfo.INT_TYPE_INFO, env.getConfig())),
+ props,
+ new KafkaPartitioner<Integer>() {
+ @Override
+ public int partition(Integer next, byte[] serializedKey, byte[] serializedValue, int numPartitions) {
+ return next % numPartitions;
+ }
+ });
env.execute("Scrambles int sequence generator");
}
// ------------------------------------------------------------------------
- public static class InfiniteStringsGenerator extends Thread {
+ public static class InfiniteStringsGenerator extends Thread implements Serializable {
- private final KafkaTestEnvironment server;
+ private transient KafkaTestEnvironment server;
private final String topic;
-
+
+ private final int flinkPort;
+
private volatile Throwable error;
private volatile boolean running = true;
- public InfiniteStringsGenerator(KafkaTestEnvironment server, String topic) {
+ public InfiniteStringsGenerator(KafkaTestEnvironment server, String topic, int flinkPort) {
this.server = server;
this.topic = topic;
+ this.flinkPort = flinkPort;
}
@Override
@@ -180,28 +183,36 @@ public class DataGenerators {
// we manually feed data into the Kafka sink
FlinkKafkaProducerBase<String> producer = null;
try {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
+ DataStream<String> stream = env.addSource(new SourceFunction<String>() {
+ @Override
+ public void run(SourceContext<String> ctx) throws Exception {
+ final StringBuilder bld = new StringBuilder();
+ final Random rnd = new Random();
+ while (running) {
+ bld.setLength(0);
+ int len = rnd.nextInt(100) + 1;
+ for (int i = 0; i < len; i++) {
+ bld.append((char) (rnd.nextInt(20) + 'a'));
+ }
+
+ String next = bld.toString();
+ ctx.collect(next);
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+ });
+
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(server.getBrokerConnectionString());
producerProperties.setProperty("retries", "3");
- producer = server.getProducer(topic,
+ server.produceIntoKafka(stream, topic,
new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()),
producerProperties, new FixedPartitioner<String>());
- producer.setRuntimeContext(new MockRuntimeContext(1,0));
- producer.open(new Configuration());
-
- final StringBuilder bld = new StringBuilder();
- final Random rnd = new Random();
-
- while (running) {
- bld.setLength(0);
-
- int len = rnd.nextInt(100) + 1;
- for (int i = 0; i < len; i++) {
- bld.append((char) (rnd.nextInt(20) + 'a') );
- }
-
- String next = bld.toString();
- producer.invoke(next);
- }
+ env.execute("String generator");
}
catch (Throwable t) {
this.error = t;
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/JobManagerCommunicationUtils.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/JobManagerCommunicationUtils.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/JobManagerCommunicationUtils.java
index 028045a..acdad5a 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/JobManagerCommunicationUtils.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/JobManagerCommunicationUtils.java
@@ -54,6 +54,10 @@ public class JobManagerCommunicationUtils {
}
public static void cancelCurrentJob(ActorGateway jobManager) throws Exception {
+ cancelCurrentJob(jobManager, null);
+ }
+
+ public static void cancelCurrentJob(ActorGateway jobManager, String name) throws Exception {
JobStatusMessage status = null;
for (int i = 0; i < 200; i++) {
@@ -78,8 +82,21 @@ public class JobManagerCommunicationUtils {
else if (jobs.size() == 1) {
status = jobs.get(0);
}
- else {
- throw new Exception("Could not cancel job - more than one running job.");
+ else if(name != null) {
+ for(JobStatusMessage msg: jobs) {
+ if(msg.getJobName().equals(name)) {
+ status = msg;
+ }
+ }
+ if(status == null) {
+ throw new Exception("Could not cancel job - no job matched expected name = '" + name +"' in " + jobs);
+ }
+ } else {
+ String jobNames = "";
+ for(JobStatusMessage jsm: jobs) {
+ jobNames += jsm.getJobName() + ", ";
+ }
+ throw new Exception("Could not cancel job - more than one running job: " + jobNames);
}
}
[2/3] flink git commit: [FLINK-4035] Refactor the Kafka 0.10
connector to be based upon the 0.9 connector
Posted by rm...@apache.org.
[FLINK-4035] Refactor the Kafka 0.10 connector to be based upon the 0.9 connector
Add a test case for Kafka's new timestamp functionality and update the documentation.
This closes #2369
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6731ec1e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6731ec1e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6731ec1e
Branch: refs/heads/master
Commit: 6731ec1e48d0a0092dd2330adda73bcf37fda8d7
Parents: 63859c6
Author: Robert Metzger <rm...@apache.org>
Authored: Tue Aug 9 16:38:21 2016 +0200
Committer: Robert Metzger <rm...@apache.org>
Committed: Tue Oct 11 10:04:25 2016 +0200
----------------------------------------------------------------------
docs/dev/connectors/kafka.md | 67 +++-
docs/page/js/flink.js | 3 +-
.../flink-connector-kafka-0.10/pom.xml | 50 +--
.../connectors/kafka/FlinkKafkaConsumer010.java | 121 +------
.../connectors/kafka/FlinkKafkaProducer010.java | 315 +++++++++++++++++--
.../kafka/Kafka010JsonTableSource.java | 2 +-
.../connectors/kafka/Kafka010TableSource.java | 2 +-
.../kafka/internal/Kafka010Fetcher.java | 268 ++--------------
.../connectors/kafka/Kafka010ITCase.java | 266 +++++++++++-----
.../connectors/kafka/KafkaProducerTest.java | 119 -------
.../kafka/KafkaTestEnvironmentImpl.java | 80 ++++-
.../kafka/internals/SimpleConsumerThread.java | 2 +-
.../kafka/KafkaTestEnvironmentImpl.java | 7 +-
.../connectors/kafka/FlinkKafkaConsumer09.java | 4 +-
.../connectors/kafka/FlinkKafkaProducer09.java | 2 +-
.../kafka/internal/Kafka09Fetcher.java | 22 +-
.../kafka/KafkaTestEnvironmentImpl.java | 6 +-
.../kafka/FlinkKafkaConsumerBase.java | 4 +
.../kafka/FlinkKafkaProducerBase.java | 4 +-
.../kafka/internals/AbstractFetcher.java | 43 +--
...picPartitionStateWithPeriodicWatermarks.java | 4 +-
...cPartitionStateWithPunctuatedWatermarks.java | 4 +-
.../connectors/kafka/KafkaConsumerTestBase.java | 201 ++++++------
.../connectors/kafka/KafkaProducerTestBase.java | 5 +-
.../kafka/KafkaShortRetentionTestBase.java | 4 +-
.../connectors/kafka/KafkaTestEnvironment.java | 7 +-
.../AbstractFetcherTimestampsTest.java | 68 ++--
.../kafka/testutils/DataGenerators.java | 87 ++---
.../testutils/JobManagerCommunicationUtils.java | 21 +-
29 files changed, 936 insertions(+), 852 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/docs/dev/connectors/kafka.md
----------------------------------------------------------------------
diff --git a/docs/dev/connectors/kafka.md b/docs/dev/connectors/kafka.md
index d2221fa..9a360d4 100644
--- a/docs/dev/connectors/kafka.md
+++ b/docs/dev/connectors/kafka.md
@@ -46,14 +46,6 @@ For most users, the `FlinkKafkaConsumer08` (part of `flink-connector-kafka`) is
</thead>
<tbody>
<tr>
- <td>flink-connector-kafka</td>
- <td>0.9.1, 0.10</td>
- <td>FlinkKafkaConsumer082<br>
- FlinkKafkaProducer</td>
- <td>0.8.x</td>
- <td>Uses the <a href="https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example">SimpleConsumer</a> API of Kafka internally. Offsets are committed to ZK by Flink.</td>
- </tr>
- <tr>
<td>flink-connector-kafka-0.8{{ site.scala_version_suffix }}</td>
<td>1.0.0</td>
<td>FlinkKafkaConsumer08<br>
@@ -61,7 +53,7 @@ For most users, the `FlinkKafkaConsumer08` (part of `flink-connector-kafka`) is
<td>0.8.x</td>
<td>Uses the <a href="https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example">SimpleConsumer</a> API of Kafka internally. Offsets are committed to ZK by Flink.</td>
</tr>
- <tr>
+ <tr>
<td>flink-connector-kafka-0.9{{ site.scala_version_suffix }}</td>
<td>1.0.0</td>
<td>FlinkKafkaConsumer09<br>
@@ -69,6 +61,14 @@ For most users, the `FlinkKafkaConsumer08` (part of `flink-connector-kafka`) is
<td>0.9.x</td>
<td>Uses the new <a href="http://kafka.apache.org/documentation.html#newconsumerapi">Consumer API</a> Kafka.</td>
</tr>
+ <tr>
+ <td>flink-connector-kafka-0.10{{ site.scala_version_suffix }}</td>
+ <td>1.2.0</td>
+ <td>FlinkKafkaConsumer010<br>
+ FlinkKafkaProducer010</td>
+ <td>0.10.x</td>
+ <td>This connector supports <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message">Kafka messages with timestamps</a> both for producing and consuming.</td>
+ </tr>
</tbody>
</table>
@@ -87,7 +87,6 @@ Note that the streaming connectors are currently not part of the binary distribu
### Installing Apache Kafka
* Follow the instructions from [Kafka's quickstart](https://kafka.apache.org/documentation.html#quickstart) to download the code and launch a server (launching a Zookeeper and a Kafka server is required every time before starting the application).
-* On 32 bit computers [this](http://stackoverflow.com/questions/22325364/unrecognized-vm-option-usecompressedoops-when-running-kafka-from-my-ubuntu-in) problem may occur.
* If the Kafka and Zookeeper servers are running on a remote machine, then the `advertised.host.name` setting in the `config/server.properties` file must be set to the machine's IP address.
### Kafka Consumer
@@ -256,17 +255,28 @@ records to partitions.
Example:
+
<div class="codetabs" markdown="1">
-<div data-lang="java" markdown="1">
+<div data-lang="java, Kafka 0.8+" markdown="1">
{% highlight java %}
stream.addSink(new FlinkKafkaProducer08<String>("localhost:9092", "my-topic", new SimpleStringSchema()));
{% endhighlight %}
</div>
-<div data-lang="scala" markdown="1">
+<div data-lang="java, Kafka 0.10+" markdown="1">
+{% highlight java %}
+FlinkKafkaProducer010.writeToKafkaWithTimestamps(stream, "my-topic", new SimpleStringSchema(), properties);
+{% endhighlight %}
+</div>
+<div data-lang="scala, Kafka 0.8+" markdown="1">
{% highlight scala %}
stream.addSink(new FlinkKafkaProducer08[String]("localhost:9092", "my-topic", new SimpleStringSchema()))
{% endhighlight %}
</div>
+<div data-lang="scala, Kafka 0.10+" markdown="1">
+{% highlight scala %}
+FlinkKafkaProducer010.writeToKafkaWithTimestamps(stream, "my-topic", new SimpleStringSchema(), properties);
+{% endhighlight %}
+</div>
</div>
You can also define a custom Kafka producer configuration for the KafkaSink with the constructor. Please refer to
@@ -287,3 +297,36 @@ higher value.
There is currently no transactional producer for Kafka, so Flink can not guarantee exactly-once delivery
into a Kafka topic.
+
+### Using Kafka timestamps and Flink event time in Kafka 0.10
+
+Since Apache Kafka 0.10., Kafka's messages can carry [timestamps](https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message), indicating
+the time the event has occurred (see ["event time" in Apache Flink](../event_time.html)) or the time when the message
+has been written to the Kafka broker.
+
+The `FlinkKafkaConsumer010` will emit records with the timestamp attached, if the time characteristic in Flink is
+set to `TimeCharacteristic.EventTime` (`StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)`).
+
+The Kafka consumer does not emit watermarks. To emit watermarks, the same mechanisms as described above in
+"Kafka Consumers and Timestamp Extraction/Watermark Emission" using the `assignTimestampsAndWatermarks` method are applicable.
+
+There is no need to define a timestamp extractor when using the timestamps from Kafka. The `previousElementTimestamp` argument of
+the `extractTimestamp()` method contains the timestamp carried by the Kafka message.
+
+A timestamp extractor for a Kafka consumer would look like this:
+{% highlight java %}
+public long extractTimestamp(Long element, long previousElementTimestamp) {
+ return previousElementTimestamp;
+}
+{% endhighlight %}
+
+
+
+The `FlinkKafkaProducer010` only emits the record timestamp, if `setWriteTimestampToKafka(true)` is set.
+
+{% highlight java %}
+FlinkKafkaProducer010.FlinkKafkaProducer010Configuration config = FlinkKafkaProducer010.writeToKafkaWithTimestamps(streamWithTimestamps, topic, new SimpleStringSchema(), standardProps);
+config.setWriteTimestampToKafka(true);
+{% endhighlight %}
+
+
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/docs/page/js/flink.js
----------------------------------------------------------------------
diff --git a/docs/page/js/flink.js b/docs/page/js/flink.js
index fdf972c..885a8ff 100644
--- a/docs/page/js/flink.js
+++ b/docs/page/js/flink.js
@@ -42,6 +42,7 @@ function codeTabs() {
var image = $(this).data("image");
var notabs = $(this).data("notabs");
var capitalizedLang = lang.substr(0, 1).toUpperCase() + lang.substr(1);
+ lang = lang.replace(/[^a-zA-Z0-9]/g, "_");
var id = "tab_" + lang + "_" + counter;
$(this).attr("id", id);
if (image != null && langImages[lang]) {
@@ -99,9 +100,7 @@ function viewSolution() {
// A script to fix internal hash links because we have an overlapping top bar.
// Based on https://github.com/twitter/bootstrap/issues/193#issuecomment-2281510
function maybeScrollToHash() {
- console.log("HERE");
if (window.location.hash && $(window.location.hash).length) {
- console.log("HERE2", $(window.location.hash), $(window.location.hash).offset().top);
var newTop = $(window.location.hash).offset().top - 57;
$(window).scrollTop(newTop);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml b/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml
index f2bcb11..0b426b5 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml
@@ -26,7 +26,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-connectors</artifactId>
- <version>1.1-SNAPSHOT</version>
+ <version>1.2-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
@@ -37,7 +37,7 @@ under the License.
<!-- Allow users to pass custom connector versions -->
<properties>
- <kafka.version>0.10.0.0</kafka.version>
+ <kafka.version>0.10.0.1</kafka.version>
</properties>
<dependencies>
@@ -46,21 +46,16 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_2.10</artifactId>
+ <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
<version>${project.version}</version>
- <scope>provided</scope>
</dependency>
+ <!-- Add Kafka 0.10.x as a dependency -->
+
<dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kafka-base_2.10</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka_${scala.binary.version}</artifactId>
- </exclusion>
- </exclusions>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
</dependency>
<dependency>
@@ -73,20 +68,29 @@ under the License.
<optional>true</optional>
</dependency>
+ <!-- test dependencies -->
+
<dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <version>${kafka.version}</version>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <!-- exclude Kafka dependencies -->
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.binary.version}</artifactId>
+ </exclusion>
+ </exclusions>
+ <type>test-jar</type>
+ <scope>test</scope>
</dependency>
- <!-- test dependencies -->
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-base_2.10</artifactId>
<version>${project.version}</version>
<exclusions>
- <!-- exclude 0.8 dependencies -->
+ <!-- exclude Kafka dependencies -->
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_${scala.binary.version}</artifactId>
@@ -127,6 +131,14 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics-jmx</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
index 78ccd4a..267ff57 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
@@ -28,20 +28,10 @@ import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper;
import org.apache.flink.util.SerializedValue;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
-import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* The Flink Kafka Consumer is a streaming data source that pulls a parallel data stream from
@@ -64,30 +54,10 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* 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 FlinkKafkaConsumer010<T> extends FlinkKafkaConsumerBase<T> {
+public class FlinkKafkaConsumer010<T> extends FlinkKafkaConsumer09<T> {
private static final long serialVersionUID = 2324564345203409112L;
- private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaConsumer010.class);
-
- /** Configuration key to change the polling timeout **/
- public static final String KEY_POLL_TIMEOUT = "flink.poll-timeout";
-
- /** Boolean configuration key to disable metrics tracking **/
- public static final String KEY_DISABLE_METRICS = "flink.disable-metrics";
-
- /** From Kafka's Javadoc: The time, in milliseconds, spent waiting in poll if data is not
- * available. If 0, returns immediately with any records that are available now. */
- public static final long DEFAULT_POLL_TIMEOUT = 100L;
-
- // ------------------------------------------------------------------------
-
- /** User-supplied properties for Kafka **/
- private final Properties properties;
-
- /** From Kafka's Javadoc: The time, in milliseconds, spent waiting in poll if data is not
- * available. If 0, returns immediately with any records that are available now */
- private final long pollTimeout;
// ------------------------------------------------------------------------
@@ -151,51 +121,7 @@ public class FlinkKafkaConsumer010<T> extends FlinkKafkaConsumerBase<T> {
* The properties that are used to configure both the fetcher and the offset handler.
*/
public FlinkKafkaConsumer010(List<String> topics, KeyedDeserializationSchema<T> deserializer, Properties props) {
- super(deserializer);
-
- checkNotNull(topics, "topics");
- this.properties = checkNotNull(props, "props");
- setDeserializer(this.properties);
-
- // configure the polling timeout
- try {
- if (properties.containsKey(KEY_POLL_TIMEOUT)) {
- this.pollTimeout = Long.parseLong(properties.getProperty(KEY_POLL_TIMEOUT));
- } else {
- this.pollTimeout = DEFAULT_POLL_TIMEOUT;
- }
- }
- catch (Exception e) {
- throw new IllegalArgumentException("Cannot parse poll timeout for '" + KEY_POLL_TIMEOUT + '\'', e);
- }
-
- // read the partitions that belong to the listed topics
- final List<KafkaTopicPartition> partitions = new ArrayList<>();
-
- try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(this.properties)) {
- for (final String topic: topics) {
- // get partitions for each topic
- List<PartitionInfo> partitionsForTopic = consumer.partitionsFor(topic);
- // for non existing topics, the list might be null.
- if (partitionsForTopic != null) {
- partitions.addAll(convertToFlinkKafkaTopicPartition(partitionsForTopic));
- }
- }
- }
-
- if (partitions.isEmpty()) {
- throw new RuntimeException("Unable to retrieve any partitions for the requested topics " + topics);
- }
-
- // we now have a list of partitions which is the same for all parallel consumer instances.
- LOG.info("Got {} partitions from these topics: {}", partitions.size(), topics);
-
- if (LOG.isInfoEnabled()) {
- logPartitionInfo(LOG, partitions);
- }
-
- // register these partitions
- setSubscribedPartitions(partitions);
+ super(topics, deserializer, props);
}
@Override
@@ -212,48 +138,5 @@ public class FlinkKafkaConsumer010<T> extends FlinkKafkaConsumerBase<T> {
watermarksPeriodic, watermarksPunctuated,
runtimeContext, deserializer,
properties, pollTimeout, useMetrics);
-
- }
-
- // ------------------------------------------------------------------------
- // Utilities
- // ------------------------------------------------------------------------
-
- /**
- * Converts a list of Kafka PartitionInfo's to Flink's KafkaTopicPartition (which are serializable)
- *
- * @param partitions A list of Kafka PartitionInfos.
- * @return A list of KafkaTopicPartitions
- */
- private static List<KafkaTopicPartition> convertToFlinkKafkaTopicPartition(List<PartitionInfo> partitions) {
- checkNotNull(partitions);
-
- List<KafkaTopicPartition> ret = new ArrayList<>(partitions.size());
- for (PartitionInfo pi : partitions) {
- ret.add(new KafkaTopicPartition(pi.topic(), pi.partition()));
- }
- return ret;
- }
-
- /**
- * Makes sure that the ByteArrayDeserializer is registered in the Kafka properties.
- *
- * @param props The Kafka properties to register the serializer in.
- */
- private static void setDeserializer(Properties props) {
- final String deSerName = ByteArrayDeserializer.class.getCanonicalName();
-
- Object keyDeSer = props.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
- Object valDeSer = props.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
-
- if (keyDeSer != null && !keyDeSer.equals(deSerName)) {
- LOG.warn("Ignoring configured key DeSerializer ({})", ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
- }
- if (valDeSer != null && !valDeSer.equals(deSerName)) {
- LOG.warn("Ignoring configured value DeSerializer ({})", ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
- }
-
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, deSerName);
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deSerName);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
index 49bce39..cc0194b 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
@@ -17,27 +17,123 @@
package org.apache.flink.streaming.connectors.kafka;
+import org.apache.flink.api.common.functions.IterationRuntimeContext;
+import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.StreamSink;
import org.apache.flink.streaming.connectors.kafka.partitioner.FixedPartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
+import static org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.getPropertiesFromBrokerList;
+
/**
- * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.8.
+ * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.10.x
+ *
+ * Implementation note: This producer is a hybrid between a regular regular sink function (a)
+ * and a custom operator (b).
+ *
+ * For (a), the class implements the SinkFunction and RichFunction interfaces.
+ * For (b), it extends the StreamTask class.
+ *
+ * Details about approach (a):
*
- * Please note that this producer does not have any reliability guarantees.
+ * Pre Kafka 0.10 producers only follow approach (a), allowing users to use the producer using the
+ * DataStream.addSink() method.
+ * Since the APIs exposed in that variant do not allow accessing the the timestamp attached to the record
+ * the Kafka 0.10 producer has a second invocation option, approach (b).
*
- * @param <IN> Type of the messages to write into Kafka.
+ * Details about approach (b):
+ * Kafka 0.10 supports writing the timestamp attached to a record to Kafka. When adding the
+ * FlinkKafkaProducer010 using the FlinkKafkaProducer010.writeToKafkaWithTimestamps() method, the Kafka producer
+ * can access the internal record timestamp of the record and write it to Kafka.
+ *
+ * All methods and constructors in this class are marked with the approach they are needed for.
*/
-public class FlinkKafkaProducer010<IN> extends FlinkKafkaProducerBase<IN> {
+public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunction<T>, RichFunction {
+
+ /**
+ * Flag controlling whether we are writing the Flink record's timestamp into Kafka.
+ */
+ private boolean writeTimestampToKafka = false;
+
+ // ---------------------- "Constructors" for timestamp writing ------------------
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * This constructor allows writing timestamps to Kafka, it follow approach (b) (see above)
+ *
+ * @param inStream The stream to write to Kafka
+ * @param topicId ID of the Kafka topic.
+ * @param serializationSchema User defined serialization schema supporting key/value messages
+ * @param producerConfig Properties with the producer configuration.
+ */
+ public static <T> FlinkKafkaProducer010Configuration<T> writeToKafkaWithTimestamps(DataStream<T> inStream,
+ String topicId,
+ KeyedSerializationSchema<T> serializationSchema,
+ Properties producerConfig) {
+ return writeToKafkaWithTimestamps(inStream, topicId, serializationSchema, producerConfig, new FixedPartitioner<T>());
+ }
+
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. the sink produces a DataStream to
+ * the topic.
+ *
+ * This constructor allows writing timestamps to Kafka, it follow approach (b) (see above)
+ *
+ * @param inStream The stream to write to Kafka
+ * @param topicId ID of the Kafka topic.
+ * @param serializationSchema User defined (keyless) serialization schema.
+ * @param producerConfig Properties with the producer configuration.
+ */
+ public static <T> FlinkKafkaProducer010Configuration<T> writeToKafkaWithTimestamps(DataStream<T> inStream,
+ String topicId,
+ SerializationSchema<T> serializationSchema,
+ Properties producerConfig) {
+ return writeToKafkaWithTimestamps(inStream, topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, new FixedPartitioner<T>());
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * This constructor allows writing timestamps to Kafka, it follow approach (b) (see above)
+ *
+ * @param inStream The stream to write to Kafka
+ * @param topicId The name of the target topic
+ * @param serializationSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
+ * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
+ * @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions.
+ */
+ public static <T> FlinkKafkaProducer010Configuration<T> writeToKafkaWithTimestamps(DataStream<T> inStream,
+ String topicId,
+ KeyedSerializationSchema<T> serializationSchema,
+ Properties producerConfig,
+ KafkaPartitioner<T> customPartitioner) {
- private static final long serialVersionUID = 1L;
+ GenericTypeInfo<Object> objectTypeInfo = new GenericTypeInfo<>(Object.class);
+ FlinkKafkaProducer010<T> kafkaProducer = new FlinkKafkaProducer010<>(topicId, serializationSchema, producerConfig, customPartitioner);
+ SingleOutputStreamOperator<Object> transformation = inStream.transform("FlinKafkaProducer 0.10.x", objectTypeInfo, kafkaProducer);
+ return new FlinkKafkaProducer010Configuration<>(transformation, kafkaProducer);
+ }
- // ------------------- Keyless serialization schema constructors ----------------------
+ // ---------------------- Regular constructors w/o timestamp support ------------------
/**
* Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
@@ -50,8 +146,8 @@ public class FlinkKafkaProducer010<IN> extends FlinkKafkaProducerBase<IN> {
* @param serializationSchema
* User defined (keyless) serialization schema.
*/
- public FlinkKafkaProducer010(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) {
- this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), getPropertiesFromBrokerList(brokerList), new FixedPartitioner<IN>());
+ public FlinkKafkaProducer010(String brokerList, String topicId, SerializationSchema<T> serializationSchema) {
+ this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), getPropertiesFromBrokerList(brokerList), new FixedPartitioner<T>());
}
/**
@@ -65,8 +161,8 @@ public class FlinkKafkaProducer010<IN> extends FlinkKafkaProducerBase<IN> {
* @param producerConfig
* Properties with the producer configuration.
*/
- public FlinkKafkaProducer010(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) {
- this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, new FixedPartitioner<IN>());
+ public FlinkKafkaProducer010(String topicId, SerializationSchema<T> serializationSchema, Properties producerConfig) {
+ this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, new FixedPartitioner<T>());
}
/**
@@ -78,9 +174,8 @@ public class FlinkKafkaProducer010<IN> extends FlinkKafkaProducerBase<IN> {
* @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
* @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions (when passing null, we'll use Kafka's partitioner)
*/
- public FlinkKafkaProducer010(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner<IN> customPartitioner) {
+ public FlinkKafkaProducer010(String topicId, SerializationSchema<T> serializationSchema, Properties producerConfig, KafkaPartitioner<T> customPartitioner) {
this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, customPartitioner);
-
}
// ------------------- Key/Value serialization schema constructors ----------------------
@@ -96,8 +191,8 @@ public class FlinkKafkaProducer010<IN> extends FlinkKafkaProducerBase<IN> {
* @param serializationSchema
* User defined serialization schema supporting key/value messages
*/
- public FlinkKafkaProducer010(String brokerList, String topicId, KeyedSerializationSchema<IN> serializationSchema) {
- this(topicId, serializationSchema, getPropertiesFromBrokerList(brokerList), new FixedPartitioner<IN>());
+ public FlinkKafkaProducer010(String brokerList, String topicId, KeyedSerializationSchema<T> serializationSchema) {
+ this(topicId, serializationSchema, getPropertiesFromBrokerList(brokerList), new FixedPartitioner<T>());
}
/**
@@ -111,27 +206,193 @@ public class FlinkKafkaProducer010<IN> extends FlinkKafkaProducerBase<IN> {
* @param producerConfig
* Properties with the producer configuration.
*/
- public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig) {
- this(topicId, serializationSchema, producerConfig, new FixedPartitioner<IN>());
+ public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<T> serializationSchema, Properties producerConfig) {
+ this(topicId, serializationSchema, producerConfig, new FixedPartitioner<T>());
}
/**
- * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
- * the topic.
+ * Create Kafka producer
*
- * @param topicId The topic to write data to
- * @param serializationSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
- * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
- * @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions.
+ * This constructor does not allow writing timestamps to Kafka, it follow approach (a) (see above)
*/
- public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner<IN> customPartitioner) {
- super(topicId, serializationSchema, producerConfig, customPartitioner);
+ public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<T> serializationSchema, Properties producerConfig, KafkaPartitioner<T> customPartitioner) {
+ // We create a Kafka 09 producer instance here and only "override" (by intercepting) the
+ // invoke call.
+ super(new FlinkKafkaProducer09<>(topicId, serializationSchema, producerConfig, customPartitioner));
+ }
+
+
+ // ----------------------------- Generic element processing ---------------------------
+
+ private void invokeInternal(T next, long elementTimestamp) throws Exception {
+
+ final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
+
+ internalProducer.checkErroneous();
+
+ byte[] serializedKey = internalProducer.schema.serializeKey(next);
+ byte[] serializedValue = internalProducer.schema.serializeValue(next);
+ String targetTopic = internalProducer.schema.getTargetTopic(next);
+ if (targetTopic == null) {
+ targetTopic = internalProducer.defaultTopicId;
+ }
+
+ Long timestamp = null;
+ if(this.writeTimestampToKafka) {
+ timestamp = elementTimestamp;
+ }
+
+ ProducerRecord<byte[], byte[]> record;
+ if (internalProducer.partitioner == null) {
+ record = new ProducerRecord<>(targetTopic, null, timestamp, serializedKey, serializedValue);
+ } else {
+ record = new ProducerRecord<>(targetTopic, internalProducer.partitioner.partition(next, serializedKey, serializedValue, internalProducer.partitions.length), timestamp, serializedKey, serializedValue);
+ }
+ if (internalProducer.flushOnCheckpoint) {
+ synchronized (internalProducer.pendingRecordsLock) {
+ internalProducer.pendingRecords++;
+ }
+ }
+ internalProducer.producer.send(record, internalProducer.callback);
}
+
+ // ----------------- Helper methods implementing methods from SinkFunction and RichFunction (Approach (a)) ----
+
+
+ // ---- Configuration setters
+
+ /**
+ * Defines whether the producer should fail on errors, or only log them.
+ * If this is set to true, then exceptions will be only logged, if set to false,
+ * exceptions will be eventually thrown and cause the streaming program to
+ * fail (and enter recovery).
+ *
+ * Method is only accessible for approach (a) (see above)
+ *
+ * @param logFailuresOnly The flag to indicate logging-only on exceptions.
+ */
+ public void setLogFailuresOnly(boolean logFailuresOnly) {
+ final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
+ internalProducer.setLogFailuresOnly(logFailuresOnly);
+ }
+
+ /**
+ * If set to true, the Flink producer will wait for all outstanding messages in the Kafka buffers
+ * to be acknowledged by the Kafka producer on a checkpoint.
+ * This way, the producer can guarantee that messages in the Kafka buffers are part of the checkpoint.
+ *
+ * Method is only accessible for approach (a) (see above)
+ *
+ * @param flush Flag indicating the flushing mode (true = flush on checkpoint)
+ */
+ public void setFlushOnCheckpoint(boolean flush) {
+ final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
+ internalProducer.setFlushOnCheckpoint(flush);
+ }
+
+ /**
+ * This method is used for approach (a) (see above)
+ *
+ */
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
+ internalProducer.open(parameters);
+ }
+
+ /**
+ * This method is used for approach (a) (see above)
+ */
+ @Override
+ public IterationRuntimeContext getIterationRuntimeContext() {
+ final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
+ return internalProducer.getIterationRuntimeContext();
+ }
+
+ /**
+ * This method is used for approach (a) (see above)
+ */
@Override
- protected void flush() {
- if (this.producer != null) {
- producer.flush();
+ public void setRuntimeContext(RuntimeContext t) {
+ final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
+ internalProducer.setRuntimeContext(t);
+ }
+
+ /**
+ * Invoke method for using the Sink as DataStream.addSink() sink.
+ *
+ * This method is used for approach (a) (see above)
+ *
+ * @param value The input record.
+ */
+ @Override
+ public void invoke(T value) throws Exception {
+ invokeInternal(value, Long.MAX_VALUE);
+ }
+
+
+ // ----------------- Helper methods and classes implementing methods from StreamSink (Approach (b)) ----
+
+
+ /**
+ * Process method for using the sink with timestamp support.
+ *
+ * This method is used for approach (b) (see above)
+ */
+ @Override
+ public void processElement(StreamRecord<T> element) throws Exception {
+ invokeInternal(element.getValue(), element.getTimestamp());
+ }
+
+ /**
+ * Configuration object returned by the writeToKafkaWithTimestamps() call.
+ */
+ public static class FlinkKafkaProducer010Configuration<T> extends DataStreamSink<T> {
+
+ private final FlinkKafkaProducerBase wrappedProducerBase;
+ private final FlinkKafkaProducer010 producer;
+
+ private FlinkKafkaProducer010Configuration(DataStream stream, FlinkKafkaProducer010<T> producer) {
+ //noinspection unchecked
+ super(stream, producer);
+ this.producer = producer;
+ this.wrappedProducerBase = (FlinkKafkaProducerBase) producer.userFunction;
+ }
+
+ /**
+ * Defines whether the producer should fail on errors, or only log them.
+ * If this is set to true, then exceptions will be only logged, if set to false,
+ * exceptions will be eventually thrown and cause the streaming program to
+ * fail (and enter recovery).
+ *
+ * @param logFailuresOnly The flag to indicate logging-only on exceptions.
+ */
+ public void setLogFailuresOnly(boolean logFailuresOnly) {
+ this.wrappedProducerBase.setLogFailuresOnly(logFailuresOnly);
+ }
+
+ /**
+ * If set to true, the Flink producer will wait for all outstanding messages in the Kafka buffers
+ * to be acknowledged by the Kafka producer on a checkpoint.
+ * This way, the producer can guarantee that messages in the Kafka buffers are part of the checkpoint.
+ *
+ * @param flush Flag indicating the flushing mode (true = flush on checkpoint)
+ */
+ public void setFlushOnCheckpoint(boolean flush) {
+ this.wrappedProducerBase.setFlushOnCheckpoint(flush);
+ }
+
+ /**
+ * If set to true, Flink will write the (event time) timestamp attached to each record into Kafka.
+ * Timestamps must be positive for Kafka to accept them.
+ *
+ * @param writeTimestampToKafka Flag indicating if Flink's internal timestamps are written to Kafka.
+ */
+ public void setWriteTimestampToKafka(boolean writeTimestampToKafka) {
+ this.producer.writeTimestampToKafka = writeTimestampToKafka;
}
}
+
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
index cda68ce..ddf1ad3 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
@@ -28,7 +28,7 @@ import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-public class Kafka010JsonTableSource extends KafkaJsonTableSource {
+public class Kafka010JsonTableSource extends Kafka09JsonTableSource {
/**
* Creates a Kafka 0.10 JSON {@link StreamTableSource}.
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
index cee1b90..732440b 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
@@ -28,7 +28,7 @@ import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-public class Kafka010TableSource extends KafkaTableSource {
+public class Kafka010TableSource extends Kafka09TableSource {
/**
* Creates a Kafka 0.10 {@link StreamTableSource}.
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
index 70f530b..47bee22 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
@@ -18,37 +18,20 @@
package org.apache.flink.streaming.connectors.kafka.internal;
-import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
-import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
-import org.apache.flink.streaming.connectors.kafka.internals.ExceptionProxy;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionState;
-import org.apache.flink.streaming.connectors.kafka.internals.metrics.DefaultKafkaMetricAccumulator;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.util.SerializedValue;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.common.Metric;
-import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.WakeupException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Properties;
/**
@@ -56,40 +39,7 @@ import java.util.Properties;
*
* @param <T> The type of elements produced by the fetcher.
*/
-public class Kafka010Fetcher<T> extends AbstractFetcher<T, TopicPartition> implements Runnable {
-
- private static final Logger LOG = LoggerFactory.getLogger(Kafka010Fetcher.class);
-
- // ------------------------------------------------------------------------
-
- /** The schema to convert between Kafka's byte messages, and Flink's objects */
- private final KeyedDeserializationSchema<T> deserializer;
-
- /** The subtask's runtime context */
- private final RuntimeContext runtimeContext;
-
- /** The configuration for the Kafka consumer */
- private final Properties kafkaProperties;
-
- /** The maximum number of milliseconds to wait for a fetch batch */
- private final long pollTimeout;
-
- /** Flag whether to register Kafka metrics as Flink accumulators */
- private final boolean forwardKafkaMetrics;
-
- /** Mutex to guard against concurrent access to the non-threadsafe Kafka consumer */
- private final Object consumerLock = new Object();
-
- /** Reference to the Kafka consumer, once it is created */
- private volatile KafkaConsumer<byte[], byte[]> consumer;
-
- /** Reference to the proxy, forwarding exceptions from the fetch thread to the main thread */
- private volatile ExceptionProxy errorHandler;
-
- /** Flag to mark the main work loop as alive */
- private volatile boolean running = true;
-
- // ------------------------------------------------------------------------
+public class Kafka010Fetcher<T> extends Kafka09Fetcher<T> {
public Kafka010Fetcher(
SourceContext<T> sourceContext,
@@ -100,213 +50,47 @@ public class Kafka010Fetcher<T> extends AbstractFetcher<T, TopicPartition> imple
KeyedDeserializationSchema<T> deserializer,
Properties kafkaProperties,
long pollTimeout,
- boolean forwardKafkaMetrics) throws Exception
+ boolean useMetrics) throws Exception
{
- super(sourceContext, assignedPartitions, watermarksPeriodic, watermarksPunctuated, runtimeContext);
-
- this.deserializer = deserializer;
- this.runtimeContext = runtimeContext;
- this.kafkaProperties = kafkaProperties;
- this.pollTimeout = pollTimeout;
- this.forwardKafkaMetrics = forwardKafkaMetrics;
-
- // if checkpointing is enabled, we are not automatically committing to Kafka.
- kafkaProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
- Boolean.toString(!runtimeContext.isCheckpointingEnabled()));
+ super(sourceContext, assignedPartitions, watermarksPeriodic, watermarksPunctuated, runtimeContext, deserializer, kafkaProperties, pollTimeout, useMetrics);
}
- // ------------------------------------------------------------------------
- // Fetcher work methods
- // ------------------------------------------------------------------------
-
@Override
- public void runFetchLoop() throws Exception {
- this.errorHandler = new ExceptionProxy(Thread.currentThread());
-
- // rather than running the main fetch loop directly here, we spawn a dedicated thread
- // this makes sure that no interrupt() call upon canceling reaches the Kafka consumer code
- Thread runner = new Thread(this, "Kafka 0.10 Fetcher for " + runtimeContext.getTaskNameWithSubtasks());
- runner.setDaemon(true);
- runner.start();
-
- try {
- runner.join();
- } catch (InterruptedException e) {
- // may be the result of a wake-up after an exception. we ignore this here and only
- // restore the interruption state
- Thread.currentThread().interrupt();
- }
-
- // make sure we propagate any exception that occurred in the concurrent fetch thread,
- // before leaving this method
- this.errorHandler.checkAndThrowException();
+ protected void assignPartitionsToConsumer(KafkaConsumer<byte[], byte[]> consumer, List<TopicPartition> topicPartitions) {
+ consumer.assign(topicPartitions);
}
@Override
- public void cancel() {
- // flag the main thread to exit
- running = false;
-
- // NOTE:
- // - We cannot interrupt the runner thread, because the Kafka consumer may
- // deadlock when the thread is interrupted while in certain methods
- // - We cannot call close() on the consumer, because it will actually throw
- // an exception if a concurrent call is in progress
-
- // make sure the consumer finds out faster that we are shutting down
- if (consumer != null) {
- consumer.wakeup();
- }
+ protected void emitRecord(T record, KafkaTopicPartitionState<TopicPartition> partition, long offset, ConsumerRecord consumerRecord) throws Exception {
+ // get timestamp from provided ConsumerRecord (only possible with kafka 0.10.x)
+ super.emitRecord(record, partition, offset, consumerRecord.timestamp());
}
+ /**
+ * Emit record Kafka-timestamp aware.
+ */
@Override
- public void run() {
- // This method initializes the KafkaConsumer and guarantees it is torn down properly.
- // This is important, because the consumer has multi-threading issues,
- // including concurrent 'close()' calls.
-
- final KafkaConsumer<byte[], byte[]> consumer;
- try {
- consumer = new KafkaConsumer<>(kafkaProperties);
- }
- catch (Throwable t) {
- running = false;
- errorHandler.reportError(t);
- return;
- }
-
- // from here on, the consumer will be closed properly
- try {
- consumer.assign(convertKafkaPartitions(subscribedPartitions()));
-
- // register Kafka metrics to Flink accumulators
- if (forwardKafkaMetrics) {
- Map<MetricName, ? extends Metric> metrics = consumer.metrics();
- if (metrics == null) {
- // MapR's Kafka implementation returns null here.
- LOG.info("Consumer implementation does not support metrics");
- } else {
- // we have metrics, register them where possible
- for (Map.Entry<MetricName, ? extends Metric> metric : metrics.entrySet()) {
- String name = "KafkaConsumer-" + metric.getKey().name();
- DefaultKafkaMetricAccumulator kafkaAccumulator =
- DefaultKafkaMetricAccumulator.createFor(metric.getValue());
-
- // best effort: we only add the accumulator if available.
- if (kafkaAccumulator != null) {
- runtimeContext.addAccumulator(name, kafkaAccumulator);
- }
- }
- }
- }
-
- // seek the consumer to the initial offsets
- for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitions()) {
- if (partition.isOffsetDefined()) {
- consumer.seek(partition.getKafkaPartitionHandle(), partition.getOffset() + 1);
- }
- }
-
- // from now on, external operations may call the consumer
- this.consumer = consumer;
-
- // main fetch loop
- while (running) {
- // get the next batch of records
- final ConsumerRecords<byte[], byte[]> records;
- synchronized (consumerLock) {
- try {
- records = consumer.poll(pollTimeout);
- }
- catch (WakeupException we) {
- if (running) {
- throw we;
- } else {
- continue;
- }
- }
- }
-
- // get the records for each topic partition
- for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitions()) {
-
- List<ConsumerRecord<byte[], byte[]>> partitionRecords = records.records(partition.getKafkaPartitionHandle());
-
- for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
- T value = deserializer.deserialize(
- record.key(), record.value(),
- record.topic(), record.partition(), record.offset());
-
- if (deserializer.isEndOfStream(value)) {
- // end of stream signaled
- running = false;
- break;
- }
-
- // emit the actual record. this also update offset state atomically
- // and deals with timestamps and watermark generation
- emitRecord(value, partition, record.offset());
- }
- }
+ protected void emitRecord(T record, KafkaTopicPartitionState<TopicPartition> partitionState, long offset, long timestamp) throws Exception {
+ if (timestampWatermarkMode == NO_TIMESTAMPS_WATERMARKS) {
+ // fast path logic, in case there are no watermarks
+
+ // emit the record, using the checkpoint lock to guarantee
+ // atomicity of record emission and offset state update
+ synchronized (checkpointLock) {
+ sourceContext.collectWithTimestamp(record, timestamp);
+ partitionState.setOffset(offset);
}
- // end main fetch loop
}
- catch (Throwable t) {
- if (running) {
- running = false;
- errorHandler.reportError(t);
- } else {
- LOG.debug("Stopped ConsumerThread threw exception", t);
- }
+ else if (timestampWatermarkMode == PERIODIC_WATERMARKS) {
+ emitRecordWithTimestampAndPeriodicWatermark(record, partitionState, offset, timestamp);
}
- finally {
- try {
- synchronized (consumerLock) {
- consumer.close();
- }
- } catch (Throwable t) {
- LOG.warn("Error while closing Kafka 0.10 consumer", t);
- }
+ else {
+ emitRecordWithTimestampAndPunctuatedWatermark(record, partitionState, offset, timestamp);
}
}
- // ------------------------------------------------------------------------
- // Kafka 0.10 specific fetcher behavior
- // ------------------------------------------------------------------------
-
@Override
- public TopicPartition createKafkaPartitionHandle(KafkaTopicPartition partition) {
- return new TopicPartition(partition.getTopic(), partition.getPartition());
- }
-
- @Override
- public void commitSpecificOffsetsToKafka(Map<KafkaTopicPartition, Long> offsets) throws Exception {
- KafkaTopicPartitionState<TopicPartition>[] partitions = subscribedPartitions();
- Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>(partitions.length);
-
- for (KafkaTopicPartitionState<TopicPartition> partition : partitions) {
- Long offset = offsets.get(partition.getKafkaTopicPartition());
- if (offset != null) {
- offsetsToCommit.put(partition.getKafkaPartitionHandle(), new OffsetAndMetadata(offset, ""));
- }
- }
-
- if (this.consumer != null) {
- synchronized (consumerLock) {
- this.consumer.commitSync(offsetsToCommit);
- }
- }
- }
-
- // ------------------------------------------------------------------------
- // Utilities
- // ------------------------------------------------------------------------
-
- public static Collection<TopicPartition> convertKafkaPartitions(KafkaTopicPartitionState<TopicPartition>[] partitions) {
- ArrayList<TopicPartition> result = new ArrayList<>(partitions.length);
- for (KafkaTopicPartitionState<TopicPartition> p : partitions) {
- result.add(p.getKafkaPartitionHandle());
- }
- return result;
+ protected String getFetcherName() {
+ return "Kafka 0.10 Fetcher";
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
index 5427853..28bf6d5 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
@@ -17,14 +17,32 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
+import org.apache.flink.streaming.util.serialization.TypeInformationSerializationSchema;
import org.junit.Test;
-import java.util.UUID;
-
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import javax.annotation.Nullable;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
public class Kafka010ITCase extends KafkaConsumerTestBase {
@@ -33,10 +51,6 @@ public class Kafka010ITCase extends KafkaConsumerTestBase {
// Suite of Tests
// ------------------------------------------------------------------------
- @Override
- public String getExpectedKafkaVersion() {
- return "0.10";
- }
@Test(timeout = 60000)
public void testFailOnNoBroker() throws Exception {
@@ -48,16 +62,6 @@ public class Kafka010ITCase extends KafkaConsumerTestBase {
runSimpleConcurrentProducerConsumerTopology();
}
-// @Test(timeout = 60000)
-// public void testPunctuatedExplicitWMConsumer() throws Exception {
-// runExplicitPunctuatedWMgeneratingConsumerTest(false);
-// }
-
-// @Test(timeout = 60000)
-// public void testPunctuatedExplicitWMConsumerWithEmptyTopic() throws Exception {
-// runExplicitPunctuatedWMgeneratingConsumerTest(true);
-// }
-
@Test(timeout = 60000)
public void testKeyValueSupport() throws Exception {
runKeyValueTest();
@@ -124,68 +128,168 @@ public class Kafka010ITCase extends KafkaConsumerTestBase {
@Test(timeout = 60000)
public void testMetricsAndEndOfStream() throws Exception {
- runMetricsAndEndOfStreamTest();
- }
-
- @Test
- public void testJsonTableSource() throws Exception {
- String topic = UUID.randomUUID().toString();
-
- // Names and types are determined in the actual test method of the
- // base test class.
- Kafka010JsonTableSource tableSource = new Kafka010JsonTableSource(
- topic,
- standardProps,
- new String[] {
- "long",
- "string",
- "boolean",
- "double",
- "missing-field"},
- new TypeInformation<?>[] {
- BasicTypeInfo.LONG_TYPE_INFO,
- BasicTypeInfo.STRING_TYPE_INFO,
- BasicTypeInfo.BOOLEAN_TYPE_INFO,
- BasicTypeInfo.DOUBLE_TYPE_INFO,
- BasicTypeInfo.LONG_TYPE_INFO });
-
- // Don't fail on missing field, but set to null (default)
- tableSource.setFailOnMissingField(false);
-
- runJsonTableSource(topic, tableSource);
- }
-
- @Test
- public void testJsonTableSourceWithFailOnMissingField() throws Exception {
- String topic = UUID.randomUUID().toString();
-
- // Names and types are determined in the actual test method of the
- // base test class.
- Kafka010JsonTableSource tableSource = new Kafka010JsonTableSource(
- topic,
- standardProps,
- new String[] {
- "long",
- "string",
- "boolean",
- "double",
- "missing-field"},
- new TypeInformation<?>[] {
- BasicTypeInfo.LONG_TYPE_INFO,
- BasicTypeInfo.STRING_TYPE_INFO,
- BasicTypeInfo.BOOLEAN_TYPE_INFO,
- BasicTypeInfo.DOUBLE_TYPE_INFO,
- BasicTypeInfo.LONG_TYPE_INFO });
-
- // Don't fail on missing field, but set to null (default)
- tableSource.setFailOnMissingField(true);
-
- try {
- runJsonTableSource(topic, tableSource);
- fail("Did not throw expected Exception");
- } catch (Exception e) {
- Throwable rootCause = e.getCause().getCause().getCause();
- assertTrue("Unexpected root cause", rootCause instanceof IllegalStateException);
+ runEndOfStreamTest();
+ }
+
+ /**
+ * Kafka 0.10 specific test, ensuring Timestamps are properly written to and read from Kafka
+ */
+ @Test(timeout = 60000)
+ public void testTimestamps() throws Exception {
+
+ final String topic = "tstopic";
+ createTestTopic(topic, 3, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
+ env.setParallelism(1);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.getConfig().disableSysoutLogging();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+ DataStream<Long> streamWithTimestamps = env.addSource(new SourceFunction<Long>() {
+ boolean running = true;
+
+ @Override
+ public void run(SourceContext<Long> ctx) throws Exception {
+ long i = 0;
+ while(running) {
+ ctx.collectWithTimestamp(i, i*2);
+ if(i++ == 1000L) {
+ running = false;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+ });
+
+ final TypeInformationSerializationSchema<Long> longSer = new TypeInformationSerializationSchema<>(TypeInfoParser.<Long>parse("Long"), env.getConfig());
+ FlinkKafkaProducer010.FlinkKafkaProducer010Configuration prod = FlinkKafkaProducer010.writeToKafkaWithTimestamps(streamWithTimestamps, topic, new KeyedSerializationSchemaWrapper<>(longSer), standardProps, new KafkaPartitioner<Long>() {
+ @Override
+ public int partition(Long next, byte[] serializedKey, byte[] serializedValue, int numPartitions) {
+ return (int)(next % 3);
+ }
+ });
+ prod.setParallelism(3);
+ prod.setWriteTimestampToKafka(true);
+ env.execute("Produce some");
+
+ // ---------- Consume stream from Kafka -------------------
+
+ env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
+ env.setParallelism(1);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.getConfig().disableSysoutLogging();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+ FlinkKafkaConsumer010<Long> kafkaSource = new FlinkKafkaConsumer010<>(topic, new LimitedLongDeserializer(), standardProps);
+ kafkaSource.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<Long>() {
+ @Nullable
+ @Override
+ public Watermark checkAndGetNextWatermark(Long lastElement, long extractedTimestamp) {
+ if(lastElement % 10 == 0) {
+ return new Watermark(lastElement);
+ }
+ return null;
+ }
+
+ @Override
+ public long extractTimestamp(Long element, long previousElementTimestamp) {
+ return previousElementTimestamp;
+ }
+ });
+
+ DataStream<Long> stream = env.addSource(kafkaSource);
+ GenericTypeInfo<Object> objectTypeInfo = new GenericTypeInfo<>(Object.class);
+ stream.transform("timestamp validating operator", objectTypeInfo, new TimestampValidatingOperator()).setParallelism(1);
+
+ env.execute("Consume again");
+
+ deleteTestTopic(topic);
+ }
+
+ private static class TimestampValidatingOperator extends StreamSink<Long> {
+
+ public TimestampValidatingOperator() {
+ super(new SinkFunction<Long>() {
+ @Override
+ public void invoke(Long value) throws Exception {
+ throw new RuntimeException("Unexpected");
+ }
+ });
+ }
+
+ long elCount = 0;
+ long wmCount = 0;
+ long lastWM = Long.MIN_VALUE;
+
+ @Override
+ public void processElement(StreamRecord<Long> element) throws Exception {
+ elCount++;
+ if(element.getValue() * 2 != element.getTimestamp()) {
+ throw new RuntimeException("Invalid timestamp: " + element);
+ }
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ wmCount++;
+
+ if(lastWM <= mark.getTimestamp()) {
+ lastWM = mark.getTimestamp();
+ } else {
+ throw new RuntimeException("Received watermark higher than the last one");
+ }
+
+ if( mark.getTimestamp() % 10 != 0 && mark.getTimestamp() != Long.MAX_VALUE ) {
+ throw new RuntimeException("Invalid watermark: " + mark.getTimestamp());
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ if(elCount != 1000L) {
+ throw new RuntimeException("Wrong final element count " + elCount);
+ }
+
+ if(wmCount <= 2) {
+ throw new RuntimeException("Almost no watermarks have been sent " + wmCount);
+ }
+ }
+ }
+
+ private static class LimitedLongDeserializer implements KeyedDeserializationSchema<Long> {
+
+ private final TypeInformation<Long> ti;
+ private final TypeSerializer<Long> ser;
+ long cnt = 0;
+
+ public LimitedLongDeserializer() {
+ this.ti = TypeInfoParser.parse("Long");
+ this.ser = ti.createSerializer(new ExecutionConfig());
+ }
+ @Override
+ public TypeInformation<Long> getProducedType() {
+ return ti;
+ }
+
+ @Override
+ public Long deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
+ cnt++;
+ DataInputView in = new DataInputViewStreamWrapper(new ByteArrayInputStream(message));
+ Long e = ser.deserialize(in);
+ return e;
+ }
+
+ @Override
+ public boolean isEndOfStream(Long nextElement) {
+ return cnt > 1000L;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
deleted file mode 100644
index 5f5ac63..0000000
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
+++ /dev/null
@@ -1,119 +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.kafka;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.connectors.kafka.testutils.MockRuntimeContext;
-import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
-import org.apache.flink.util.TestLogger;
-
-import org.apache.kafka.clients.producer.Callback;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.common.PartitionInfo;
-
-import org.junit.Test;
-import org.junit.runner.RunWith;
-
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
-
-import java.util.Collections;
-import java.util.Properties;
-import java.util.concurrent.Future;
-
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyString;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.powermock.api.mockito.PowerMockito.whenNew;
-
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(FlinkKafkaProducerBase.class)
-public class KafkaProducerTest extends TestLogger {
-
- @Test
- @SuppressWarnings("unchecked")
- public void testPropagateExceptions() {
- try {
- // mock kafka producer
- KafkaProducer<?, ?> kafkaProducerMock = mock(KafkaProducer.class);
-
- // partition setup
- when(kafkaProducerMock.partitionsFor(anyString())).thenReturn(
- Collections.singletonList(new PartitionInfo("mock_topic", 42, null, null, null)));
-
- // failure when trying to send an element
- when(kafkaProducerMock.send(any(ProducerRecord.class), any(Callback.class)))
- .thenAnswer(new Answer<Future<RecordMetadata>>() {
- @Override
- public Future<RecordMetadata> answer(InvocationOnMock invocation) throws Throwable {
- Callback callback = (Callback) invocation.getArguments()[1];
- callback.onCompletion(null, new Exception("Test error"));
- return null;
- }
- });
-
- // make sure the FlinkKafkaProducer instantiates our mock producer
- whenNew(KafkaProducer.class).withAnyArguments().thenReturn(kafkaProducerMock);
-
- // (1) producer that propagates errors
-
- FlinkKafkaProducer010<String> producerPropagating = new FlinkKafkaProducer010<>(
- "mock_topic", new SimpleStringSchema(), new Properties(), null);
-
- producerPropagating.setRuntimeContext(new MockRuntimeContext(17, 3));
- producerPropagating.open(new Configuration());
-
- try {
- producerPropagating.invoke("value");
- producerPropagating.invoke("value");
- fail("This should fail with an exception");
- }
- catch (Exception e) {
- assertNotNull(e.getCause());
- assertNotNull(e.getCause().getMessage());
- assertTrue(e.getCause().getMessage().contains("Test error"));
- }
-
- // (2) producer that only logs errors
-
- FlinkKafkaProducer010<String> producerLogging = new FlinkKafkaProducer010<>(
- "mock_topic", new SimpleStringSchema(), new Properties(), null);
- producerLogging.setLogFailuresOnly(true);
-
- producerLogging.setRuntimeContext(new MockRuntimeContext(17, 3));
- producerLogging.open(new Configuration());
-
- producerLogging.invoke("value");
- producerLogging.invoke("value");
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index 45f0478..af6d254 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -28,6 +28,8 @@ import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.io.FileUtils;
import org.apache.curator.test.TestingServer;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.connectors.kafka.testutils.ZooKeeperStringSerializer;
import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
@@ -64,6 +66,9 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
private String brokerConnectionString = "";
private Properties standardProps;
private Properties additionalServerProperties;
+ private boolean secureMode = false;
+ // 6 seconds is default. Seems to be too small for travis. 30 seconds
+ private int zkTimeout = 30000;
public String getBrokerConnectionString() {
return brokerConnectionString;
@@ -75,6 +80,22 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
+ public Properties getSecureProperties() {
+ Properties prop = new Properties();
+ if(secureMode) {
+ prop.put("security.inter.broker.protocol", "SASL_PLAINTEXT");
+ prop.put("security.protocol", "SASL_PLAINTEXT");
+ prop.put("sasl.kerberos.service.name", "kafka");
+
+ //add special timeout for Travis
+ prop.setProperty("zookeeper.session.timeout.ms", String.valueOf(zkTimeout));
+ prop.setProperty("zookeeper.connection.timeout.ms", String.valueOf(zkTimeout));
+ prop.setProperty("metadata.fetch.timeout.ms","120000");
+ }
+ return prop;
+ }
+
+ @Override
public String getVersion() {
return "0.10";
}
@@ -90,10 +111,13 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> FlinkKafkaProducerBase<T> getProducer(String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) {
+ public <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) {
FlinkKafkaProducer010<T> prod = new FlinkKafkaProducer010<>(topic, serSchema, props, partitioner);
prod.setFlushOnCheckpoint(true);
- return prod;
+ return stream.addSink(prod);
+ /* FlinkKafkaProducer010.FlinkKafkaProducer010Configuration<T> sink = FlinkKafkaProducer010.writeToKafkaWithTimestamps(stream, topic, serSchema, props, partitioner);
+ sink.setFlushOnCheckpoint(true);
+ return sink; */
}
@Override
@@ -130,8 +154,21 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public void prepare(int numKafkaServers, Properties additionalServerProperties) {
+ public boolean isSecureRunSupported() {
+ return true;
+ }
+
+ @Override
+ public void prepare(int numKafkaServers, Properties additionalServerProperties, boolean secureMode) {
+ //increase the timeout since in Travis ZK connection takes long time for secure connection.
+ if(secureMode) {
+ //run only one kafka server to avoid multiple ZK connections from many instances - Travis timeout
+ numKafkaServers = 1;
+ zkTimeout = zkTimeout * 15;
+ }
+
this.additionalServerProperties = additionalServerProperties;
+ this.secureMode = secureMode;
File tempDir = new File(System.getProperty("java.io.tmpdir"));
tmpZkDir = new File(tempDir, "kafkaITcase-zk-dir-" + (UUID.randomUUID().toString()));
@@ -151,9 +188,9 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
brokers = null;
try {
- LOG.info("Starting Zookeeper");
- zookeeper = new TestingServer(-1, tmpZkDir);
+ zookeeper = new TestingServer(- 1, tmpZkDir);
zookeeperConnectionString = zookeeper.getConnectString();
+ LOG.info("Starting Zookeeper with zookeeperConnectionString: {}", zookeeperConnectionString);
LOG.info("Starting KafkaServer");
brokers = new ArrayList<>(numKafkaServers);
@@ -161,8 +198,11 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
for (int i = 0; i < numKafkaServers; i++) {
brokers.add(getKafkaServer(i, tmpKafkaDirs.get(i)));
- SocketServer socketServer = brokers.get(i).socketServer();
- brokerConnectionString += hostAndPortToUrlString(KafkaTestEnvironment.KAFKA_HOST, brokers.get(i).socketServer().boundPort(SecurityProtocol.PLAINTEXT)) + ",";
+ if(secureMode) {
+ brokerConnectionString += hostAndPortToUrlString(KafkaTestEnvironment.KAFKA_HOST, brokers.get(i).socketServer().boundPort(SecurityProtocol.SASL_PLAINTEXT)) + ",";
+ } else {
+ brokerConnectionString += hostAndPortToUrlString(KafkaTestEnvironment.KAFKA_HOST, brokers.get(i).socketServer().boundPort(SecurityProtocol.PLAINTEXT)) + ",";
+ }
}
LOG.info("ZK and KafkaServer started.");
@@ -177,8 +217,8 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
standardProps.setProperty("bootstrap.servers", brokerConnectionString);
standardProps.setProperty("group.id", "flink-tests");
standardProps.setProperty("auto.commit.enable", "false");
- standardProps.setProperty("zookeeper.session.timeout.ms", "30000"); // 6 seconds is default. Seems to be too small for travis.
- standardProps.setProperty("zookeeper.connection.timeout.ms", "30000");
+ standardProps.setProperty("zookeeper.session.timeout.ms", String.valueOf(zkTimeout));
+ standardProps.setProperty("zookeeper.connection.timeout.ms", String.valueOf(zkTimeout));
standardProps.setProperty("auto.offset.reset", "earliest"); // read from the beginning. (earliest is kafka 0.10 value)
standardProps.setProperty("fetch.message.max.bytes", "256"); // make a lot of fetches (MESSAGES MUST BE SMALLER!)
}
@@ -244,7 +284,14 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
final long deadline = System.currentTimeMillis() + 30000;
do {
try {
- Thread.sleep(100);
+ if(secureMode) {
+ //increase wait time since in Travis ZK timeout occurs frequently
+ int wait = zkTimeout / 100;
+ LOG.info("waiting for {} msecs before the topic {} can be checked", wait, topic);
+ Thread.sleep(wait);
+ } else {
+ Thread.sleep(100);
+ }
} catch (InterruptedException e) {
// restore interrupted state
}
@@ -295,8 +342,8 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
kafkaProperties.put("replica.fetch.max.bytes", String.valueOf(50 * 1024 * 1024));
// for CI stability, increase zookeeper session timeout
- kafkaProperties.put("zookeeper.session.timeout.ms", "30000");
- kafkaProperties.put("zookeeper.connection.timeout.ms", "30000");
+ kafkaProperties.put("zookeeper.session.timeout.ms", zkTimeout);
+ kafkaProperties.put("zookeeper.connection.timeout.ms", zkTimeout);
if(additionalServerProperties != null) {
kafkaProperties.putAll(additionalServerProperties);
}
@@ -306,6 +353,15 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
for (int i = 1; i <= numTries; i++) {
int kafkaPort = NetUtils.getAvailablePort();
kafkaProperties.put("port", Integer.toString(kafkaPort));
+
+ //to support secure kafka cluster
+ if(secureMode) {
+ LOG.info("Adding Kafka secure configurations");
+ kafkaProperties.put("listeners", "SASL_PLAINTEXT://" + KAFKA_HOST + ":" + kafkaPort);
+ kafkaProperties.put("advertised.listeners", "SASL_PLAINTEXT://" + KAFKA_HOST + ":" + kafkaPort);
+ kafkaProperties.putAll(getSecureProperties());
+ }
+
KafkaConfig kafkaConfig = new KafkaConfig(kafkaProperties);
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
index 35e491a..1302348 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
@@ -376,7 +376,7 @@ class SimpleConsumerThread<T> extends Thread {
continue partitionsLoop;
}
- owner.emitRecord(value, currentPartition, offset);
+ owner.emitRecord(value, currentPartition, offset, Long.MIN_VALUE);
}
else {
// no longer running
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index cbf3d06..a0d5002 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -31,6 +31,9 @@ import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionLeader;
import org.apache.flink.streaming.connectors.kafka.testutils.ZooKeeperStringSerializer;
import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
@@ -101,10 +104,10 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> FlinkKafkaProducerBase<T> getProducer(String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) {
+ public <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) {
FlinkKafkaProducer08<T> prod = new FlinkKafkaProducer08<>(topic, serSchema, props, partitioner);
prod.setFlushOnCheckpoint(true);
- return prod;
+ return stream.addSink(prod);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
index 9708777..a97476a 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
@@ -81,11 +81,11 @@ public class FlinkKafkaConsumer09<T> extends FlinkKafkaConsumerBase<T> {
// ------------------------------------------------------------------------
/** User-supplied properties for Kafka **/
- private final Properties properties;
+ protected final Properties properties;
/** From Kafka's Javadoc: The time, in milliseconds, spent waiting in poll if data is not
* available. If 0, returns immediately with any records that are available now */
- private final long pollTimeout;
+ protected final long pollTimeout;
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer09.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer09.java b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer09.java
index eb3440a..2a3e39d 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer09.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer09.java
@@ -27,7 +27,7 @@ import java.util.Properties;
/**
- * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.8.
+ * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.9.
*
* Please note that this producer does not have any reliability guarantees.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/6731ec1e/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
index aaec9dc..37e40fc 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
@@ -131,7 +131,7 @@ public class Kafka09Fetcher<T> extends AbstractFetcher<T, TopicPartition> implem
// rather than running the main fetch loop directly here, we spawn a dedicated thread
// this makes sure that no interrupt() call upon canceling reaches the Kafka consumer code
- Thread runner = new Thread(this, "Kafka 0.9 Fetcher for " + runtimeContext.getTaskNameWithSubtasks());
+ Thread runner = new Thread(this, getFetcherName() + " for " + runtimeContext.getTaskNameWithSubtasks());
runner.setDaemon(true);
runner.start();
@@ -183,7 +183,8 @@ public class Kafka09Fetcher<T> extends AbstractFetcher<T, TopicPartition> implem
// from here on, the consumer will be closed properly
try {
- consumer.assign(convertKafkaPartitions(subscribedPartitions()));
+ assignPartitionsToConsumer(consumer, convertKafkaPartitions(subscribedPartitions()));
+
if (useMetrics) {
final MetricGroup kafkaMetricGroup = runtimeContext.getMetricGroup().addGroup("KafkaConsumer");
@@ -250,7 +251,7 @@ public class Kafka09Fetcher<T> extends AbstractFetcher<T, TopicPartition> implem
// emit the actual record. this also update offset state atomically
// and deals with timestamps and watermark generation
- emitRecord(value, partition, record.offset());
+ emitRecord(value, partition, record.offset(), record);
}
}
}
@@ -274,6 +275,21 @@ public class Kafka09Fetcher<T> extends AbstractFetcher<T, TopicPartition> implem
}
}
+ // Kafka09Fetcher ignores the timestamp, Kafka010Fetcher is extracting the timestamp and passing it to the emitRecord() method.
+ protected void emitRecord(T record, KafkaTopicPartitionState<TopicPartition> partition, long offset, ConsumerRecord consumerRecord) throws Exception {
+ emitRecord(record, partition, offset, Long.MIN_VALUE);
+ }
+ /**
+ * Protected method to make the partition assignment pluggable, for different Kafka versions.
+ */
+ protected void assignPartitionsToConsumer(KafkaConsumer<byte[], byte[]> consumer, List<TopicPartition> topicPartitions) {
+ consumer.assign(topicPartitions);
+ }
+
+ protected String getFetcherName() {
+ return "Kafka 0.9 Fetcher";
+ }
+
// ------------------------------------------------------------------------
// Kafka 0.9 specific fetcher behavior
// ------------------------------------------------------------------------
[3/3] flink git commit: [FLINK-4035] Add support for Kafka 0.10.x.
Posted by rm...@apache.org.
[FLINK-4035] Add support for Kafka 0.10.x.
This closes #2231
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/63859c64
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/63859c64
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/63859c64
Branch: refs/heads/master
Commit: 63859c648a2a0ff024228f6e0687d837b8896322
Parents: a079259
Author: radekg <ra...@gruchalski.com>
Authored: Tue Jul 12 13:19:01 2016 -0400
Committer: Robert Metzger <rm...@apache.org>
Committed: Tue Oct 11 10:04:25 2016 +0200
----------------------------------------------------------------------
.../flink-connector-kafka-0.10/pom.xml | 179 ++++++++++
.../connectors/kafka/FlinkKafkaConsumer010.java | 259 +++++++++++++++
.../connectors/kafka/FlinkKafkaProducer010.java | 137 ++++++++
.../kafka/Kafka010JsonTableSource.java | 71 ++++
.../connectors/kafka/Kafka010TableSource.java | 75 +++++
.../kafka/internal/Kafka010Fetcher.java | 312 +++++++++++++++++
.../src/main/resources/log4j.properties | 29 ++
.../connectors/kafka/Kafka010ITCase.java | 192 +++++++++++
.../kafka/Kafka010ProducerITCase.java | 33 ++
.../connectors/kafka/KafkaProducerTest.java | 119 +++++++
.../kafka/KafkaShortRetention010ITCase.java | 34 ++
.../kafka/KafkaTestEnvironmentImpl.java | 331 +++++++++++++++++++
.../src/test/resources/log4j-test.properties | 30 ++
.../src/test/resources/logback-test.xml | 30 ++
flink-streaming-connectors/pom.xml | 1 +
15 files changed, 1832 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml b/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml
new file mode 100644
index 0000000..f2bcb11
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml
@@ -0,0 +1,179 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-connectors</artifactId>
+ <version>1.1-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-connector-kafka-0.10_2.10</artifactId>
+ <name>flink-connector-kafka-0.10</name>
+
+ <packaging>jar</packaging>
+
+ <!-- Allow users to pass custom connector versions -->
+ <properties>
+ <kafka.version>0.10.0.0</kafka.version>
+ </properties>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_2.10</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-base_2.10</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.binary.version}</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_2.10</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <!-- Projects depending on this project,
+ won't depend on flink-table. -->
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
+ </dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-base_2.10</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <!-- exclude 0.8 dependencies -->
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.binary.version}</artifactId>
+ </exclusion>
+ </exclusions>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <!-- include 0.10 server for tests -->
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.binary.version}</artifactId>
+ <version>${kafka.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-tests_2.10</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils_2.10</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_2.10</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ <configuration>
+ <includes>
+ <include>**/KafkaTestEnvironmentImpl*</include>
+ </includes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-source-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-test-sources</id>
+ <goals>
+ <goal>test-jar-no-fork</goal>
+ </goals>
+ <configuration>
+ <includes>
+ <include>**/KafkaTestEnvironmentImpl*</include>
+ </includes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <!-- Enforce single fork execution due to heavy mini cluster use in the tests -->
+ <forkCount>1</forkCount>
+ <argLine>-Xms256m -Xmx1000m -Dlog4j.configuration=${log4j.configuration} -Dmvn.forkNumber=${surefire.forkNumber} -XX:-UseGCOverheadLimit</argLine>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
new file mode 100644
index 0000000..78ccd4a
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
@@ -0,0 +1,259 @@
+/*
+ * 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.kafka;
+
+import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher;
+import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper;
+import org.apache.flink.util.SerializedValue;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The Flink Kafka Consumer is a streaming data source that pulls a parallel data stream from
+ * Apache Kafka 0.10.x. 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>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>Please refer to Kafka's documentation for the available configuration properties:
+ * http://kafka.apache.org/documentation.html#newconsumerconfigs</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 FlinkKafkaConsumer010<T> extends FlinkKafkaConsumerBase<T> {
+
+ private static final long serialVersionUID = 2324564345203409112L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaConsumer010.class);
+
+ /** Configuration key to change the polling timeout **/
+ public static final String KEY_POLL_TIMEOUT = "flink.poll-timeout";
+
+ /** Boolean configuration key to disable metrics tracking **/
+ public static final String KEY_DISABLE_METRICS = "flink.disable-metrics";
+
+ /** From Kafka's Javadoc: The time, in milliseconds, spent waiting in poll if data is not
+ * available. If 0, returns immediately with any records that are available now. */
+ public static final long DEFAULT_POLL_TIMEOUT = 100L;
+
+ // ------------------------------------------------------------------------
+
+ /** User-supplied properties for Kafka **/
+ private final Properties properties;
+
+ /** From Kafka's Javadoc: The time, in milliseconds, spent waiting in poll if data is not
+ * available. If 0, returns immediately with any records that are available now */
+ private final long pollTimeout;
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a new Kafka streaming source consumer for Kafka 0.10.x
+ *
+ * @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 FlinkKafkaConsumer010(String topic, DeserializationSchema<T> valueDeserializer, Properties props) {
+ this(Collections.singletonList(topic), valueDeserializer, props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer for Kafka 0.10.x
+ *
+ * This constructor allows passing a {@see KeyedDeserializationSchema} for reading key/value
+ * pairs, offsets, and topic names from Kafka.
+ *
+ * @param topic
+ * The name of the topic that should be consumed.
+ * @param deserializer
+ * The keyed 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 FlinkKafkaConsumer010(String topic, KeyedDeserializationSchema<T> deserializer, Properties props) {
+ this(Collections.singletonList(topic), deserializer, props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer for Kafka 0.10.x
+ *
+ * This constructor allows passing multiple topics to the consumer.
+ *
+ * @param topics
+ * The Kafka topics to read from.
+ * @param deserializer
+ * The de-/serializer used to convert between Kafka's byte messages and Flink's objects.
+ * @param props
+ * The properties that are used to configure both the fetcher and the offset handler.
+ */
+ public FlinkKafkaConsumer010(List<String> topics, DeserializationSchema<T> deserializer, Properties props) {
+ this(topics, new KeyedDeserializationSchemaWrapper<>(deserializer), props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer for Kafka 0.10.x
+ *
+ * This constructor allows passing multiple topics and a key/value deserialization schema.
+ *
+ * @param topics
+ * The Kafka topics to read from.
+ * @param deserializer
+ * The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects.
+ * @param props
+ * The properties that are used to configure both the fetcher and the offset handler.
+ */
+ public FlinkKafkaConsumer010(List<String> topics, KeyedDeserializationSchema<T> deserializer, Properties props) {
+ super(deserializer);
+
+ checkNotNull(topics, "topics");
+ this.properties = checkNotNull(props, "props");
+ setDeserializer(this.properties);
+
+ // configure the polling timeout
+ try {
+ if (properties.containsKey(KEY_POLL_TIMEOUT)) {
+ this.pollTimeout = Long.parseLong(properties.getProperty(KEY_POLL_TIMEOUT));
+ } else {
+ this.pollTimeout = DEFAULT_POLL_TIMEOUT;
+ }
+ }
+ catch (Exception e) {
+ throw new IllegalArgumentException("Cannot parse poll timeout for '" + KEY_POLL_TIMEOUT + '\'', e);
+ }
+
+ // read the partitions that belong to the listed topics
+ final List<KafkaTopicPartition> partitions = new ArrayList<>();
+
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(this.properties)) {
+ for (final String topic: topics) {
+ // get partitions for each topic
+ List<PartitionInfo> partitionsForTopic = consumer.partitionsFor(topic);
+ // for non existing topics, the list might be null.
+ if (partitionsForTopic != null) {
+ partitions.addAll(convertToFlinkKafkaTopicPartition(partitionsForTopic));
+ }
+ }
+ }
+
+ if (partitions.isEmpty()) {
+ throw new RuntimeException("Unable to retrieve any partitions for the requested topics " + topics);
+ }
+
+ // we now have a list of partitions which is the same for all parallel consumer instances.
+ LOG.info("Got {} partitions from these topics: {}", partitions.size(), topics);
+
+ if (LOG.isInfoEnabled()) {
+ logPartitionInfo(LOG, partitions);
+ }
+
+ // register these partitions
+ setSubscribedPartitions(partitions);
+ }
+
+ @Override
+ protected AbstractFetcher<T, ?> createFetcher(
+ SourceContext<T> sourceContext,
+ List<KafkaTopicPartition> thisSubtaskPartitions,
+ SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
+ SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
+ StreamingRuntimeContext runtimeContext) throws Exception {
+
+ boolean useMetrics = !Boolean.valueOf(properties.getProperty(KEY_DISABLE_METRICS, "false"));
+
+ return new Kafka010Fetcher<>(sourceContext, thisSubtaskPartitions,
+ watermarksPeriodic, watermarksPunctuated,
+ runtimeContext, deserializer,
+ properties, pollTimeout, useMetrics);
+
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ /**
+ * Converts a list of Kafka PartitionInfo's to Flink's KafkaTopicPartition (which are serializable)
+ *
+ * @param partitions A list of Kafka PartitionInfos.
+ * @return A list of KafkaTopicPartitions
+ */
+ private static List<KafkaTopicPartition> convertToFlinkKafkaTopicPartition(List<PartitionInfo> partitions) {
+ checkNotNull(partitions);
+
+ List<KafkaTopicPartition> ret = new ArrayList<>(partitions.size());
+ for (PartitionInfo pi : partitions) {
+ ret.add(new KafkaTopicPartition(pi.topic(), pi.partition()));
+ }
+ return ret;
+ }
+
+ /**
+ * Makes sure that the ByteArrayDeserializer is registered in the Kafka properties.
+ *
+ * @param props The Kafka properties to register the serializer in.
+ */
+ private static void setDeserializer(Properties props) {
+ final String deSerName = ByteArrayDeserializer.class.getCanonicalName();
+
+ Object keyDeSer = props.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
+ Object valDeSer = props.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
+
+ if (keyDeSer != null && !keyDeSer.equals(deSerName)) {
+ LOG.warn("Ignoring configured key DeSerializer ({})", ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
+ }
+ if (valDeSer != null && !valDeSer.equals(deSerName)) {
+ LOG.warn("Ignoring configured value DeSerializer ({})", ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);
+ }
+
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, deSerName);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deSerName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
new file mode 100644
index 0000000..49bce39
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
@@ -0,0 +1,137 @@
+/*
+ * 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.kafka;
+
+import org.apache.flink.streaming.connectors.kafka.partitioner.FixedPartitioner;
+import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
+import org.apache.flink.streaming.util.serialization.SerializationSchema;
+
+import java.util.Properties;
+
+
+/**
+ * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.8.
+ *
+ * Please note that this producer does not have any reliability guarantees.
+ *
+ * @param <IN> Type of the messages to write into Kafka.
+ */
+public class FlinkKafkaProducer010<IN> extends FlinkKafkaProducerBase<IN> {
+
+ private static final long serialVersionUID = 1L;
+
+ // ------------------- Keyless serialization schema constructors ----------------------
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * @param brokerList
+ * Comma separated addresses of the brokers
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined (keyless) serialization schema.
+ */
+ public FlinkKafkaProducer010(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) {
+ this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), getPropertiesFromBrokerList(brokerList), new FixedPartitioner<IN>());
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. the sink produces a DataStream to
+ * the topic.
+ *
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined (keyless) serialization schema.
+ * @param producerConfig
+ * Properties with the producer configuration.
+ */
+ public FlinkKafkaProducer010(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) {
+ this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, new FixedPartitioner<IN>());
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. the sink produces a DataStream to
+ * the topic.
+ *
+ * @param topicId The topic to write data to
+ * @param serializationSchema A (keyless) serializable serialization schema for turning user objects into a kafka-consumable byte[]
+ * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
+ * @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions (when passing null, we'll use Kafka's partitioner)
+ */
+ public FlinkKafkaProducer010(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner<IN> customPartitioner) {
+ this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, customPartitioner);
+
+ }
+
+ // ------------------- Key/Value serialization schema constructors ----------------------
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * @param brokerList
+ * Comma separated addresses of the brokers
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined serialization schema supporting key/value messages
+ */
+ public FlinkKafkaProducer010(String brokerList, String topicId, KeyedSerializationSchema<IN> serializationSchema) {
+ this(topicId, serializationSchema, getPropertiesFromBrokerList(brokerList), new FixedPartitioner<IN>());
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined serialization schema supporting key/value messages
+ * @param producerConfig
+ * Properties with the producer configuration.
+ */
+ public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig) {
+ this(topicId, serializationSchema, producerConfig, new FixedPartitioner<IN>());
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * @param topicId The topic to write data to
+ * @param serializationSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
+ * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
+ * @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions.
+ */
+ public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner<IN> customPartitioner) {
+ super(topicId, serializationSchema, producerConfig, customPartitioner);
+ }
+
+ @Override
+ protected void flush() {
+ if (this.producer != null) {
+ producer.flush();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
new file mode 100644
index 0000000..cda68ce
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
@@ -0,0 +1,71 @@
+/*
+ * 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.kafka;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.table.Row;
+import org.apache.flink.api.table.sources.StreamTableSource;
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+
+import java.util.Properties;
+
+/**
+ * Kafka {@link StreamTableSource} for Kafka 0.10.
+ */
+public class Kafka010JsonTableSource extends KafkaJsonTableSource {
+
+ /**
+ * Creates a Kafka 0.10 JSON {@link StreamTableSource}.
+ *
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param fieldNames Row field names.
+ * @param fieldTypes Row field types.
+ */
+ public Kafka010JsonTableSource(
+ String topic,
+ Properties properties,
+ String[] fieldNames,
+ TypeInformation<?>[] fieldTypes) {
+
+ super(topic, properties, fieldNames, fieldTypes);
+ }
+
+ /**
+ * Creates a Kafka 0.10 JSON {@link StreamTableSource}.
+ *
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param fieldNames Row field names.
+ * @param fieldTypes Row field types.
+ */
+ public Kafka010JsonTableSource(
+ String topic,
+ Properties properties,
+ String[] fieldNames,
+ Class<?>[] fieldTypes) {
+
+ super(topic, properties, fieldNames, fieldTypes);
+ }
+
+ @Override
+ FlinkKafkaConsumerBase<Row> getKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) {
+ return new FlinkKafkaConsumer010<>(topic, deserializationSchema, properties);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
new file mode 100644
index 0000000..cee1b90
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
@@ -0,0 +1,75 @@
+/*
+ * 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.kafka;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.table.Row;
+import org.apache.flink.api.table.sources.StreamTableSource;
+import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+
+import java.util.Properties;
+
+/**
+ * Kafka {@link StreamTableSource} for Kafka 0.10.
+ */
+public class Kafka010TableSource extends KafkaTableSource {
+
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema to use for Kafka records.
+ * @param fieldNames Row field names.
+ * @param fieldTypes Row field types.
+ */
+ public Kafka010TableSource(
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema,
+ String[] fieldNames,
+ TypeInformation<?>[] fieldTypes) {
+
+ super(topic, properties, deserializationSchema, fieldNames, fieldTypes);
+ }
+
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema to use for Kafka records.
+ * @param fieldNames Row field names.
+ * @param fieldTypes Row field types.
+ */
+ public Kafka010TableSource(
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema,
+ String[] fieldNames,
+ Class<?>[] fieldTypes) {
+
+ super(topic, properties, deserializationSchema, fieldNames, fieldTypes);
+ }
+
+ @Override
+ FlinkKafkaConsumerBase<Row> getKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) {
+ return new FlinkKafkaConsumer010<>(topic, deserializationSchema, properties);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
new file mode 100644
index 0000000..70f530b
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
@@ -0,0 +1,312 @@
+/*
+ * 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.kafka.internal;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
+import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
+import org.apache.flink.streaming.connectors.kafka.internals.ExceptionProxy;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionState;
+import org.apache.flink.streaming.connectors.kafka.internals.metrics.DefaultKafkaMetricAccumulator;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.util.SerializedValue;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * A fetcher that fetches data from Kafka brokers via the Kafka 0.10 consumer API.
+ *
+ * @param <T> The type of elements produced by the fetcher.
+ */
+public class Kafka010Fetcher<T> extends AbstractFetcher<T, TopicPartition> implements Runnable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Kafka010Fetcher.class);
+
+ // ------------------------------------------------------------------------
+
+ /** The schema to convert between Kafka's byte messages, and Flink's objects */
+ private final KeyedDeserializationSchema<T> deserializer;
+
+ /** The subtask's runtime context */
+ private final RuntimeContext runtimeContext;
+
+ /** The configuration for the Kafka consumer */
+ private final Properties kafkaProperties;
+
+ /** The maximum number of milliseconds to wait for a fetch batch */
+ private final long pollTimeout;
+
+ /** Flag whether to register Kafka metrics as Flink accumulators */
+ private final boolean forwardKafkaMetrics;
+
+ /** Mutex to guard against concurrent access to the non-threadsafe Kafka consumer */
+ private final Object consumerLock = new Object();
+
+ /** Reference to the Kafka consumer, once it is created */
+ private volatile KafkaConsumer<byte[], byte[]> consumer;
+
+ /** Reference to the proxy, forwarding exceptions from the fetch thread to the main thread */
+ private volatile ExceptionProxy errorHandler;
+
+ /** Flag to mark the main work loop as alive */
+ private volatile boolean running = true;
+
+ // ------------------------------------------------------------------------
+
+ public Kafka010Fetcher(
+ SourceContext<T> sourceContext,
+ List<KafkaTopicPartition> assignedPartitions,
+ SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
+ SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
+ StreamingRuntimeContext runtimeContext,
+ KeyedDeserializationSchema<T> deserializer,
+ Properties kafkaProperties,
+ long pollTimeout,
+ boolean forwardKafkaMetrics) throws Exception
+ {
+ super(sourceContext, assignedPartitions, watermarksPeriodic, watermarksPunctuated, runtimeContext);
+
+ this.deserializer = deserializer;
+ this.runtimeContext = runtimeContext;
+ this.kafkaProperties = kafkaProperties;
+ this.pollTimeout = pollTimeout;
+ this.forwardKafkaMetrics = forwardKafkaMetrics;
+
+ // if checkpointing is enabled, we are not automatically committing to Kafka.
+ kafkaProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
+ Boolean.toString(!runtimeContext.isCheckpointingEnabled()));
+ }
+
+ // ------------------------------------------------------------------------
+ // Fetcher work methods
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void runFetchLoop() throws Exception {
+ this.errorHandler = new ExceptionProxy(Thread.currentThread());
+
+ // rather than running the main fetch loop directly here, we spawn a dedicated thread
+ // this makes sure that no interrupt() call upon canceling reaches the Kafka consumer code
+ Thread runner = new Thread(this, "Kafka 0.10 Fetcher for " + runtimeContext.getTaskNameWithSubtasks());
+ runner.setDaemon(true);
+ runner.start();
+
+ try {
+ runner.join();
+ } catch (InterruptedException e) {
+ // may be the result of a wake-up after an exception. we ignore this here and only
+ // restore the interruption state
+ Thread.currentThread().interrupt();
+ }
+
+ // make sure we propagate any exception that occurred in the concurrent fetch thread,
+ // before leaving this method
+ this.errorHandler.checkAndThrowException();
+ }
+
+ @Override
+ public void cancel() {
+ // flag the main thread to exit
+ running = false;
+
+ // NOTE:
+ // - We cannot interrupt the runner thread, because the Kafka consumer may
+ // deadlock when the thread is interrupted while in certain methods
+ // - We cannot call close() on the consumer, because it will actually throw
+ // an exception if a concurrent call is in progress
+
+ // make sure the consumer finds out faster that we are shutting down
+ if (consumer != null) {
+ consumer.wakeup();
+ }
+ }
+
+ @Override
+ public void run() {
+ // This method initializes the KafkaConsumer and guarantees it is torn down properly.
+ // This is important, because the consumer has multi-threading issues,
+ // including concurrent 'close()' calls.
+
+ final KafkaConsumer<byte[], byte[]> consumer;
+ try {
+ consumer = new KafkaConsumer<>(kafkaProperties);
+ }
+ catch (Throwable t) {
+ running = false;
+ errorHandler.reportError(t);
+ return;
+ }
+
+ // from here on, the consumer will be closed properly
+ try {
+ consumer.assign(convertKafkaPartitions(subscribedPartitions()));
+
+ // register Kafka metrics to Flink accumulators
+ if (forwardKafkaMetrics) {
+ Map<MetricName, ? extends Metric> metrics = consumer.metrics();
+ if (metrics == null) {
+ // MapR's Kafka implementation returns null here.
+ LOG.info("Consumer implementation does not support metrics");
+ } else {
+ // we have metrics, register them where possible
+ for (Map.Entry<MetricName, ? extends Metric> metric : metrics.entrySet()) {
+ String name = "KafkaConsumer-" + metric.getKey().name();
+ DefaultKafkaMetricAccumulator kafkaAccumulator =
+ DefaultKafkaMetricAccumulator.createFor(metric.getValue());
+
+ // best effort: we only add the accumulator if available.
+ if (kafkaAccumulator != null) {
+ runtimeContext.addAccumulator(name, kafkaAccumulator);
+ }
+ }
+ }
+ }
+
+ // seek the consumer to the initial offsets
+ for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitions()) {
+ if (partition.isOffsetDefined()) {
+ consumer.seek(partition.getKafkaPartitionHandle(), partition.getOffset() + 1);
+ }
+ }
+
+ // from now on, external operations may call the consumer
+ this.consumer = consumer;
+
+ // main fetch loop
+ while (running) {
+ // get the next batch of records
+ final ConsumerRecords<byte[], byte[]> records;
+ synchronized (consumerLock) {
+ try {
+ records = consumer.poll(pollTimeout);
+ }
+ catch (WakeupException we) {
+ if (running) {
+ throw we;
+ } else {
+ continue;
+ }
+ }
+ }
+
+ // get the records for each topic partition
+ for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitions()) {
+
+ List<ConsumerRecord<byte[], byte[]>> partitionRecords = records.records(partition.getKafkaPartitionHandle());
+
+ for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
+ T value = deserializer.deserialize(
+ record.key(), record.value(),
+ record.topic(), record.partition(), record.offset());
+
+ if (deserializer.isEndOfStream(value)) {
+ // end of stream signaled
+ running = false;
+ break;
+ }
+
+ // emit the actual record. this also update offset state atomically
+ // and deals with timestamps and watermark generation
+ emitRecord(value, partition, record.offset());
+ }
+ }
+ }
+ // end main fetch loop
+ }
+ catch (Throwable t) {
+ if (running) {
+ running = false;
+ errorHandler.reportError(t);
+ } else {
+ LOG.debug("Stopped ConsumerThread threw exception", t);
+ }
+ }
+ finally {
+ try {
+ synchronized (consumerLock) {
+ consumer.close();
+ }
+ } catch (Throwable t) {
+ LOG.warn("Error while closing Kafka 0.10 consumer", t);
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Kafka 0.10 specific fetcher behavior
+ // ------------------------------------------------------------------------
+
+ @Override
+ public TopicPartition createKafkaPartitionHandle(KafkaTopicPartition partition) {
+ return new TopicPartition(partition.getTopic(), partition.getPartition());
+ }
+
+ @Override
+ public void commitSpecificOffsetsToKafka(Map<KafkaTopicPartition, Long> offsets) throws Exception {
+ KafkaTopicPartitionState<TopicPartition>[] partitions = subscribedPartitions();
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>(partitions.length);
+
+ for (KafkaTopicPartitionState<TopicPartition> partition : partitions) {
+ Long offset = offsets.get(partition.getKafkaTopicPartition());
+ if (offset != null) {
+ offsetsToCommit.put(partition.getKafkaPartitionHandle(), new OffsetAndMetadata(offset, ""));
+ }
+ }
+
+ if (this.consumer != null) {
+ synchronized (consumerLock) {
+ this.consumer.commitSync(offsetsToCommit);
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ public static Collection<TopicPartition> convertKafkaPartitions(KafkaTopicPartitionState<TopicPartition>[] partitions) {
+ ArrayList<TopicPartition> result = new ArrayList<>(partitions.length);
+ for (KafkaTopicPartitionState<TopicPartition> p : partitions) {
+ result.add(p.getKafkaPartitionHandle());
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/resources/log4j.properties b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/resources/log4j.properties
new file mode 100644
index 0000000..6bdfb48
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/resources/log4j.properties
@@ -0,0 +1,29 @@
+################################################################################
+# 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/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
new file mode 100644
index 0000000..5427853
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
@@ -0,0 +1,192 @@
+/*
+ * 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.kafka;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.junit.Test;
+
+import java.util.UUID;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+
+public class Kafka010ITCase extends KafkaConsumerTestBase {
+
+ // ------------------------------------------------------------------------
+ // Suite of Tests
+ // ------------------------------------------------------------------------
+
+ @Override
+ public String getExpectedKafkaVersion() {
+ return "0.10";
+ }
+
+ @Test(timeout = 60000)
+ public void testFailOnNoBroker() throws Exception {
+ runFailOnNoBrokerTest();
+ }
+
+ @Test(timeout = 60000)
+ public void testConcurrentProducerConsumerTopology() throws Exception {
+ runSimpleConcurrentProducerConsumerTopology();
+ }
+
+// @Test(timeout = 60000)
+// public void testPunctuatedExplicitWMConsumer() throws Exception {
+// runExplicitPunctuatedWMgeneratingConsumerTest(false);
+// }
+
+// @Test(timeout = 60000)
+// public void testPunctuatedExplicitWMConsumerWithEmptyTopic() throws Exception {
+// runExplicitPunctuatedWMgeneratingConsumerTest(true);
+// }
+
+ @Test(timeout = 60000)
+ public void testKeyValueSupport() throws Exception {
+ runKeyValueTest();
+ }
+
+ // --- canceling / failures ---
+
+ @Test(timeout = 60000)
+ public void testCancelingEmptyTopic() throws Exception {
+ runCancelingOnEmptyInputTest();
+ }
+
+ @Test(timeout = 60000)
+ public void testCancelingFullTopic() throws Exception {
+ runCancelingOnFullInputTest();
+ }
+
+ @Test(timeout = 60000)
+ public void testFailOnDeploy() throws Exception {
+ runFailOnDeployTest();
+ }
+
+
+ // --- source to partition mappings and exactly once ---
+
+ @Test(timeout = 60000)
+ public void testOneToOneSources() throws Exception {
+ runOneToOneExactlyOnceTest();
+ }
+
+ @Test(timeout = 60000)
+ public void testOneSourceMultiplePartitions() throws Exception {
+ runOneSourceMultiplePartitionsExactlyOnceTest();
+ }
+
+ @Test(timeout = 60000)
+ public void testMultipleSourcesOnePartition() throws Exception {
+ runMultipleSourcesOnePartitionExactlyOnceTest();
+ }
+
+ // --- broker failure ---
+
+ @Test(timeout = 60000)
+ public void testBrokerFailure() throws Exception {
+ runBrokerFailureTest();
+ }
+
+ // --- special executions ---
+
+ @Test(timeout = 60000)
+ public void testBigRecordJob() throws Exception {
+ runBigRecordTestTopology();
+ }
+
+ @Test(timeout = 60000)
+ public void testMultipleTopics() throws Exception {
+ runProduceConsumeMultipleTopics();
+ }
+
+ @Test(timeout = 60000)
+ public void testAllDeletes() throws Exception {
+ runAllDeletesTest();
+ }
+
+ @Test(timeout = 60000)
+ public void testMetricsAndEndOfStream() throws Exception {
+ runMetricsAndEndOfStreamTest();
+ }
+
+ @Test
+ public void testJsonTableSource() throws Exception {
+ String topic = UUID.randomUUID().toString();
+
+ // Names and types are determined in the actual test method of the
+ // base test class.
+ Kafka010JsonTableSource tableSource = new Kafka010JsonTableSource(
+ topic,
+ standardProps,
+ new String[] {
+ "long",
+ "string",
+ "boolean",
+ "double",
+ "missing-field"},
+ new TypeInformation<?>[] {
+ BasicTypeInfo.LONG_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.BOOLEAN_TYPE_INFO,
+ BasicTypeInfo.DOUBLE_TYPE_INFO,
+ BasicTypeInfo.LONG_TYPE_INFO });
+
+ // Don't fail on missing field, but set to null (default)
+ tableSource.setFailOnMissingField(false);
+
+ runJsonTableSource(topic, tableSource);
+ }
+
+ @Test
+ public void testJsonTableSourceWithFailOnMissingField() throws Exception {
+ String topic = UUID.randomUUID().toString();
+
+ // Names and types are determined in the actual test method of the
+ // base test class.
+ Kafka010JsonTableSource tableSource = new Kafka010JsonTableSource(
+ topic,
+ standardProps,
+ new String[] {
+ "long",
+ "string",
+ "boolean",
+ "double",
+ "missing-field"},
+ new TypeInformation<?>[] {
+ BasicTypeInfo.LONG_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.BOOLEAN_TYPE_INFO,
+ BasicTypeInfo.DOUBLE_TYPE_INFO,
+ BasicTypeInfo.LONG_TYPE_INFO });
+
+ // Don't fail on missing field, but set to null (default)
+ tableSource.setFailOnMissingField(true);
+
+ try {
+ runJsonTableSource(topic, tableSource);
+ fail("Did not throw expected Exception");
+ } catch (Exception e) {
+ Throwable rootCause = e.getCause().getCause().getCause();
+ assertTrue("Unexpected root cause", rootCause instanceof IllegalStateException);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java
new file mode 100644
index 0000000..42b9682
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java
@@ -0,0 +1,33 @@
+/*
+ * 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.kafka;
+
+
+import org.junit.Test;
+
+
+@SuppressWarnings("serial")
+public class Kafka010ProducerITCase extends KafkaProducerTestBase {
+
+ @Test
+ public void testCustomPartitioning() {
+ runCustomPartitioningTest();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
new file mode 100644
index 0000000..5f5ac63
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.kafka;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.connectors.kafka.testutils.MockRuntimeContext;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.PartitionInfo;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.Future;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(FlinkKafkaProducerBase.class)
+public class KafkaProducerTest extends TestLogger {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPropagateExceptions() {
+ try {
+ // mock kafka producer
+ KafkaProducer<?, ?> kafkaProducerMock = mock(KafkaProducer.class);
+
+ // partition setup
+ when(kafkaProducerMock.partitionsFor(anyString())).thenReturn(
+ Collections.singletonList(new PartitionInfo("mock_topic", 42, null, null, null)));
+
+ // failure when trying to send an element
+ when(kafkaProducerMock.send(any(ProducerRecord.class), any(Callback.class)))
+ .thenAnswer(new Answer<Future<RecordMetadata>>() {
+ @Override
+ public Future<RecordMetadata> answer(InvocationOnMock invocation) throws Throwable {
+ Callback callback = (Callback) invocation.getArguments()[1];
+ callback.onCompletion(null, new Exception("Test error"));
+ return null;
+ }
+ });
+
+ // make sure the FlinkKafkaProducer instantiates our mock producer
+ whenNew(KafkaProducer.class).withAnyArguments().thenReturn(kafkaProducerMock);
+
+ // (1) producer that propagates errors
+
+ FlinkKafkaProducer010<String> producerPropagating = new FlinkKafkaProducer010<>(
+ "mock_topic", new SimpleStringSchema(), new Properties(), null);
+
+ producerPropagating.setRuntimeContext(new MockRuntimeContext(17, 3));
+ producerPropagating.open(new Configuration());
+
+ try {
+ producerPropagating.invoke("value");
+ producerPropagating.invoke("value");
+ fail("This should fail with an exception");
+ }
+ catch (Exception e) {
+ assertNotNull(e.getCause());
+ assertNotNull(e.getCause().getMessage());
+ assertTrue(e.getCause().getMessage().contains("Test error"));
+ }
+
+ // (2) producer that only logs errors
+
+ FlinkKafkaProducer010<String> producerLogging = new FlinkKafkaProducer010<>(
+ "mock_topic", new SimpleStringSchema(), new Properties(), null);
+ producerLogging.setLogFailuresOnly(true);
+
+ producerLogging.setRuntimeContext(new MockRuntimeContext(17, 3));
+ producerLogging.open(new Configuration());
+
+ producerLogging.invoke("value");
+ producerLogging.invoke("value");
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetention010ITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetention010ITCase.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetention010ITCase.java
new file mode 100644
index 0000000..1d36198
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetention010ITCase.java
@@ -0,0 +1,34 @@
+/*
+ * 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.kafka;
+
+import org.junit.Test;
+
+@SuppressWarnings("serial")
+public class KafkaShortRetention010ITCase extends KafkaShortRetentionTestBase {
+
+ @Test(timeout=60000)
+ public void testAutoOffsetReset() throws Exception {
+ runAutoOffsetResetTest();
+ }
+
+ @Test(timeout=60000)
+ public void testAutoOffsetResetNone() throws Exception {
+ runFailOnAutoOffsetResetNone();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
new file mode 100644
index 0000000..45f0478
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -0,0 +1,331 @@
+/*
+ * 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.kafka;
+
+import kafka.admin.AdminUtils;
+import kafka.common.KafkaException;
+import kafka.network.SocketServer;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServer;
+import kafka.utils.SystemTime$;
+import kafka.utils.ZkUtils;
+import org.I0Itec.zkclient.ZkClient;
+import org.apache.commons.io.FileUtils;
+import org.apache.curator.test.TestingServer;
+import org.apache.flink.streaming.connectors.kafka.testutils.ZooKeeperStringSerializer;
+import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
+import org.apache.flink.util.NetUtils;
+import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.net.BindException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.UUID;
+
+import static org.apache.flink.util.NetUtils.hostAndPortToUrlString;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * An implementation of the KafkaServerProvider for Kafka 0.10
+ */
+public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(KafkaTestEnvironmentImpl.class);
+ private File tmpZkDir;
+ private File tmpKafkaParent;
+ private List<File> tmpKafkaDirs;
+ private List<KafkaServer> brokers;
+ private TestingServer zookeeper;
+ private String zookeeperConnectionString;
+ private String brokerConnectionString = "";
+ private Properties standardProps;
+ private Properties additionalServerProperties;
+
+ public String getBrokerConnectionString() {
+ return brokerConnectionString;
+ }
+
+ @Override
+ public Properties getStandardProperties() {
+ return standardProps;
+ }
+
+ @Override
+ public String getVersion() {
+ return "0.10";
+ }
+
+ @Override
+ public List<KafkaServer> getBrokers() {
+ return brokers;
+ }
+
+ @Override
+ public <T> FlinkKafkaConsumerBase<T> getConsumer(List<String> topics, KeyedDeserializationSchema<T> readSchema, Properties props) {
+ return new FlinkKafkaConsumer010<>(topics, readSchema, props);
+ }
+
+ @Override
+ public <T> FlinkKafkaProducerBase<T> getProducer(String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) {
+ FlinkKafkaProducer010<T> prod = new FlinkKafkaProducer010<>(topic, serSchema, props, partitioner);
+ prod.setFlushOnCheckpoint(true);
+ return prod;
+ }
+
+ @Override
+ public void restartBroker(int leaderId) throws Exception {
+ brokers.set(leaderId, getKafkaServer(leaderId, tmpKafkaDirs.get(leaderId)));
+ }
+
+ @Override
+ public int getLeaderToShutDown(String topic) throws Exception {
+ ZkUtils zkUtils = getZkUtils();
+ try {
+ MetadataResponse.PartitionMetadata firstPart = null;
+ do {
+ if (firstPart != null) {
+ LOG.info("Unable to find leader. error code {}", firstPart.error().code());
+ // not the first try. Sleep a bit
+ Thread.sleep(150);
+ }
+
+ List<MetadataResponse.PartitionMetadata> partitionMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils).partitionMetadata();
+ firstPart = partitionMetadata.get(0);
+ }
+ while (firstPart.error().code() != 0);
+
+ return firstPart.leader().id();
+ } finally {
+ zkUtils.close();
+ }
+ }
+
+ @Override
+ public int getBrokerId(KafkaServer server) {
+ return server.config().brokerId();
+ }
+
+ @Override
+ public void prepare(int numKafkaServers, Properties additionalServerProperties) {
+ this.additionalServerProperties = additionalServerProperties;
+ File tempDir = new File(System.getProperty("java.io.tmpdir"));
+
+ tmpZkDir = new File(tempDir, "kafkaITcase-zk-dir-" + (UUID.randomUUID().toString()));
+ assertTrue("cannot create zookeeper temp dir", tmpZkDir.mkdirs());
+
+ tmpKafkaParent = new File(tempDir, "kafkaITcase-kafka-dir*" + (UUID.randomUUID().toString()));
+ assertTrue("cannot create kafka temp dir", tmpKafkaParent.mkdirs());
+
+ tmpKafkaDirs = new ArrayList<>(numKafkaServers);
+ for (int i = 0; i < numKafkaServers; i++) {
+ File tmpDir = new File(tmpKafkaParent, "server-" + i);
+ assertTrue("cannot create kafka temp dir", tmpDir.mkdir());
+ tmpKafkaDirs.add(tmpDir);
+ }
+
+ zookeeper = null;
+ brokers = null;
+
+ try {
+ LOG.info("Starting Zookeeper");
+ zookeeper = new TestingServer(-1, tmpZkDir);
+ zookeeperConnectionString = zookeeper.getConnectString();
+
+ LOG.info("Starting KafkaServer");
+ brokers = new ArrayList<>(numKafkaServers);
+
+ for (int i = 0; i < numKafkaServers; i++) {
+ brokers.add(getKafkaServer(i, tmpKafkaDirs.get(i)));
+
+ SocketServer socketServer = brokers.get(i).socketServer();
+ brokerConnectionString += hostAndPortToUrlString(KafkaTestEnvironment.KAFKA_HOST, brokers.get(i).socketServer().boundPort(SecurityProtocol.PLAINTEXT)) + ",";
+ }
+
+ LOG.info("ZK and KafkaServer started.");
+ }
+ catch (Throwable t) {
+ t.printStackTrace();
+ fail("Test setup failed: " + t.getMessage());
+ }
+
+ standardProps = new Properties();
+ standardProps.setProperty("zookeeper.connect", zookeeperConnectionString);
+ standardProps.setProperty("bootstrap.servers", brokerConnectionString);
+ standardProps.setProperty("group.id", "flink-tests");
+ standardProps.setProperty("auto.commit.enable", "false");
+ standardProps.setProperty("zookeeper.session.timeout.ms", "30000"); // 6 seconds is default. Seems to be too small for travis.
+ standardProps.setProperty("zookeeper.connection.timeout.ms", "30000");
+ standardProps.setProperty("auto.offset.reset", "earliest"); // read from the beginning. (earliest is kafka 0.10 value)
+ standardProps.setProperty("fetch.message.max.bytes", "256"); // make a lot of fetches (MESSAGES MUST BE SMALLER!)
+ }
+
+ @Override
+ public void shutdown() {
+ for (KafkaServer broker : brokers) {
+ if (broker != null) {
+ broker.shutdown();
+ }
+ }
+ brokers.clear();
+
+ if (zookeeper != null) {
+ try {
+ zookeeper.stop();
+ }
+ catch (Exception e) {
+ LOG.warn("ZK.stop() failed", e);
+ }
+ zookeeper = null;
+ }
+
+ // clean up the temp spaces
+
+ if (tmpKafkaParent != null && tmpKafkaParent.exists()) {
+ try {
+ FileUtils.deleteDirectory(tmpKafkaParent);
+ }
+ catch (Exception e) {
+ // ignore
+ }
+ }
+ if (tmpZkDir != null && tmpZkDir.exists()) {
+ try {
+ FileUtils.deleteDirectory(tmpZkDir);
+ }
+ catch (Exception e) {
+ // ignore
+ }
+ }
+ }
+
+ public ZkUtils getZkUtils() {
+ ZkClient creator = new ZkClient(zookeeperConnectionString, Integer.valueOf(standardProps.getProperty("zookeeper.session.timeout.ms")),
+ Integer.valueOf(standardProps.getProperty("zookeeper.connection.timeout.ms")), new ZooKeeperStringSerializer());
+ return ZkUtils.apply(creator, false);
+ }
+
+ @Override
+ public void createTestTopic(String topic, int numberOfPartitions, int replicationFactor, Properties topicConfig) {
+ // create topic with one client
+ LOG.info("Creating topic {}", topic);
+
+ ZkUtils zkUtils = getZkUtils();
+ try {
+ AdminUtils.createTopic(zkUtils, topic, numberOfPartitions, replicationFactor, topicConfig, new kafka.admin.RackAwareMode.Enforced$());
+ } finally {
+ zkUtils.close();
+ }
+
+ // validate that the topic has been created
+ final long deadline = System.currentTimeMillis() + 30000;
+ do {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ // restore interrupted state
+ }
+ // we could use AdminUtils.topicExists(zkUtils, topic) here, but it's results are
+ // not always correct.
+
+ // create a new ZK utils connection
+ ZkUtils checkZKConn = getZkUtils();
+ if(AdminUtils.topicExists(checkZKConn, topic)) {
+ checkZKConn.close();
+ return;
+ }
+ checkZKConn.close();
+ }
+ while (System.currentTimeMillis() < deadline);
+ fail("Test topic could not be created");
+ }
+
+ @Override
+ public void deleteTestTopic(String topic) {
+ ZkUtils zkUtils = getZkUtils();
+ try {
+ LOG.info("Deleting topic {}", topic);
+
+ ZkClient zk = new ZkClient(zookeeperConnectionString, Integer.valueOf(standardProps.getProperty("zookeeper.session.timeout.ms")),
+ Integer.valueOf(standardProps.getProperty("zookeeper.connection.timeout.ms")), new ZooKeeperStringSerializer());
+
+ AdminUtils.deleteTopic(zkUtils, topic);
+
+ zk.close();
+ } finally {
+ zkUtils.close();
+ }
+ }
+
+ /**
+ * Copied from com.github.sakserv.minicluster.KafkaLocalBrokerIntegrationTest (ASL licensed)
+ */
+ protected KafkaServer getKafkaServer(int brokerId, File tmpFolder) throws Exception {
+ Properties kafkaProperties = new Properties();
+
+ // properties have to be Strings
+ kafkaProperties.put("advertised.host.name", KAFKA_HOST);
+ kafkaProperties.put("broker.id", Integer.toString(brokerId));
+ kafkaProperties.put("log.dir", tmpFolder.toString());
+ kafkaProperties.put("zookeeper.connect", zookeeperConnectionString);
+ kafkaProperties.put("message.max.bytes", String.valueOf(50 * 1024 * 1024));
+ kafkaProperties.put("replica.fetch.max.bytes", String.valueOf(50 * 1024 * 1024));
+
+ // for CI stability, increase zookeeper session timeout
+ kafkaProperties.put("zookeeper.session.timeout.ms", "30000");
+ kafkaProperties.put("zookeeper.connection.timeout.ms", "30000");
+ if(additionalServerProperties != null) {
+ kafkaProperties.putAll(additionalServerProperties);
+ }
+
+ final int numTries = 5;
+
+ for (int i = 1; i <= numTries; i++) {
+ int kafkaPort = NetUtils.getAvailablePort();
+ kafkaProperties.put("port", Integer.toString(kafkaPort));
+ KafkaConfig kafkaConfig = new KafkaConfig(kafkaProperties);
+
+ try {
+ scala.Option<String> stringNone = scala.Option.apply(null);
+ KafkaServer server = new KafkaServer(kafkaConfig, SystemTime$.MODULE$, stringNone);
+ server.startup();
+ return server;
+ }
+ catch (KafkaException e) {
+ if (e.getCause() instanceof BindException) {
+ // port conflict, retry...
+ LOG.info("Port conflict when starting Kafka Broker. Retrying...");
+ }
+ else {
+ throw e;
+ }
+ }
+ }
+
+ throw new Exception("Could not start Kafka after " + numTries + " retries due to port conflicts.");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties
new file mode 100644
index 0000000..fbeb110
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties
@@ -0,0 +1,30 @@
+################################################################################
+# 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
+log4j.logger.org.apache.zookeeper=OFF, testlogger
+log4j.logger.state.change.logger=OFF, testlogger
+log4j.logger.kafka=OFF, testlogger
http://git-wip-us.apache.org/repos/asf/flink/blob/63859c64/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml
new file mode 100644
index 0000000..45b3b92
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml
@@ -0,0 +1,30 @@
+<!--
+ ~ 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/63859c64/flink-streaming-connectors/pom.xml
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/pom.xml b/flink-streaming-connectors/pom.xml
index 113109b..78e39ca 100644
--- a/flink-streaming-connectors/pom.xml
+++ b/flink-streaming-connectors/pom.xml
@@ -40,6 +40,7 @@ under the License.
<module>flink-connector-kafka-base</module>
<module>flink-connector-kafka-0.8</module>
<module>flink-connector-kafka-0.9</module>
+ <module>flink-connector-kafka-0.10</module>
<module>flink-connector-elasticsearch</module>
<module>flink-connector-elasticsearch2</module>
<module>flink-connector-rabbitmq</module>