You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2018/10/16 15:54:56 UTC
[GitHub] aljoscha closed pull request #6703: [FLINK-9697] Provide connector
for modern Kafka
aljoscha closed pull request #6703: [FLINK-9697] Provide connector for modern Kafka
URL: https://github.com/apache/flink/pull/6703
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSink.java b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSink.java
index 8471908a9cf..b9a5350e618 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSink.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSink.java
@@ -28,7 +28,7 @@
import java.util.Properties;
/**
- * Kafka 0.10 {@link KafkaTableSink} that serializes data in JSON format.
+ * Kafka 0.10 {@link KafkaTableSinkBase} that serializes data in JSON format.
*
* @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
* with descriptors for schema and format instead. Descriptors allow for
@@ -39,7 +39,7 @@
public class Kafka010JsonTableSink extends KafkaJsonTableSink {
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.10
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.10
* topic with fixed partition assignment.
*
* <p>Each parallel TableSink instance will write its rows to a single Kafka partition.</p>
@@ -60,7 +60,7 @@ public Kafka010JsonTableSink(String topic, Properties properties) {
}
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.10
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.10
* topic with custom partition assignment.
*
* @param topic topic in Kafka to which table is written
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSink.java b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSink.java
index 1d408b8ab52..79aad7c9800 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSink.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSink.java
@@ -31,7 +31,7 @@
* Kafka 0.10 table sink for writing data into Kafka.
*/
@Internal
-public class Kafka010TableSink extends KafkaTableSink {
+public class Kafka010TableSink extends KafkaTableSinkBase {
public Kafka010TableSink(
TableSchema schema,
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
index f6574623b2a..2016f84d66c 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
@@ -36,7 +36,7 @@
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
@Internal
-public class Kafka010TableSource extends KafkaTableSource {
+public class Kafka010TableSource extends KafkaTableSourceBase {
/**
* Creates a Kafka 0.10 {@link StreamTableSource}.
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactory.java b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactory.java
index ecf12b27a08..de7d7bda9a9 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactory.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactory.java
@@ -49,7 +49,7 @@ protected boolean supportsKafkaTimestamps() {
}
@Override
- protected KafkaTableSource createKafkaTableSource(
+ protected KafkaTableSourceBase createKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -73,7 +73,7 @@ protected KafkaTableSource createKafkaTableSource(
}
@Override
- protected KafkaTableSink createKafkaTableSink(
+ protected KafkaTableSinkBase createKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge010.java b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge010.java
index 5815bfade38..579e9d10893 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge010.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge010.java
@@ -35,7 +35,7 @@
* <p>Because of that, we need two versions whose compiled code goes against different method signatures.
*/
@Internal
-public class KafkaConsumerCallBridge010 extends KafkaConsumerCallBridge {
+public class KafkaConsumerCallBridge010 extends KafkaConsumerCallBridge09 {
@Override
public void assignPartitions(KafkaConsumer<?, ?> consumer, List<TopicPartition> topicPartitions) throws Exception {
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSourceTest.java
index bf253c43ece..22547af31dd 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka010AvroTableSourceTest extends KafkaAvroTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka010AvroTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java
index 9208f6583b8..4575f8763df 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java
@@ -32,10 +32,10 @@
* drop support for format-specific table sinks.
*/
@Deprecated
-public class Kafka010JsonTableSinkTest extends KafkaTableSinkTestBase {
+public class Kafka010JsonTableSinkTest extends KafkaTableSinkBaseTestBase {
@Override
- protected KafkaTableSink createTableSink(
+ protected KafkaTableSinkBase createTableSink(
String topic,
Properties properties,
FlinkKafkaPartitioner<Row> partitioner) {
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java
index 087f3edc513..330e55ad0e3 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka010JsonTableSourceTest extends KafkaJsonTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka010JsonTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactoryTest.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactoryTest.java
index dac8a4dacdd..1af001d4e40 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactoryTest.java
@@ -56,7 +56,7 @@ protected String getKafkaVersion() {
}
@Override
- protected KafkaTableSource getExpectedKafkaTableSource(
+ protected KafkaTableSourceBase getExpectedKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -81,7 +81,7 @@ protected KafkaTableSource getExpectedKafkaTableSource(
}
@Override
- protected KafkaTableSink getExpectedKafkaTableSink(
+ protected KafkaTableSinkBase getExpectedKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
index b4416e8a209..1cf4acf5406 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
@@ -38,7 +38,7 @@
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.internal.TransactionalIdsGenerator;
-import org.apache.flink.streaming.connectors.kafka.internal.metrics.KafkaMetricMuttableWrapper;
+import org.apache.flink.streaming.connectors.kafka.internal.metrics.KafkaMetricMutableWrapper;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaDelegatePartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
@@ -255,7 +255,7 @@
private final AtomicLong pendingRecords = new AtomicLong();
/** Cache of metrics to replace already registered metrics instead of overwriting existing ones. */
- private final Map<String, KafkaMetricMuttableWrapper> previouslyCreatedMetrics = new HashMap<>();
+ private final Map<String, KafkaMetricMutableWrapper> previouslyCreatedMetrics = new HashMap<>();
/**
* Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
@@ -986,12 +986,12 @@ private void recycleTransactionalProducer(FlinkKafkaProducer<byte[], byte[]> pro
String name = entry.getKey().name();
Metric metric = entry.getValue();
- KafkaMetricMuttableWrapper wrapper = previouslyCreatedMetrics.get(name);
+ KafkaMetricMutableWrapper wrapper = previouslyCreatedMetrics.get(name);
if (wrapper != null) {
wrapper.setKafkaMetric(metric);
} else {
// TODO: somehow merge metrics from all active producers?
- wrapper = new KafkaMetricMuttableWrapper(metric);
+ wrapper = new KafkaMetricMutableWrapper(metric);
previouslyCreatedMetrics.put(name, wrapper);
kafkaMetricGroup.gauge(name, wrapper);
}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSink.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSink.java
index 8d81a5b59a1..304b26ba105 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSink.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSink.java
@@ -33,7 +33,7 @@
* Kafka 0.11 table sink for writing data into Kafka.
*/
@Internal
-public class Kafka011TableSink extends KafkaTableSink {
+public class Kafka011TableSink extends KafkaTableSinkBase {
public Kafka011TableSink(
TableSchema schema,
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSource.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSource.java
index a646317fe0a..688a64972cb 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSource.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSource.java
@@ -36,7 +36,7 @@
* Kafka {@link StreamTableSource} for Kafka 0.11.
*/
@Internal
-public class Kafka011TableSource extends KafkaTableSource {
+public class Kafka011TableSource extends KafkaTableSourceBase {
/**
* Creates a Kafka 0.11 {@link StreamTableSource}.
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactory.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactory.java
index e6f677fb568..4585dc03e55 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactory.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactory.java
@@ -49,7 +49,7 @@ protected boolean supportsKafkaTimestamps() {
}
@Override
- protected KafkaTableSource createKafkaTableSource(
+ protected KafkaTableSourceBase createKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -73,7 +73,7 @@ protected KafkaTableSource createKafkaTableSource(
}
@Override
- protected KafkaTableSink createKafkaTableSink(
+ protected KafkaTableSinkBase createKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMuttableWrapper.java b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMutableWrapper.java
similarity index 90%
rename from flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMuttableWrapper.java
rename to flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMutableWrapper.java
index 3ff63636432..7d82f804577 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMuttableWrapper.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMutableWrapper.java
@@ -27,10 +27,10 @@
* Gauge for getting the current value of a Kafka metric.
*/
@Internal
-public class KafkaMetricMuttableWrapper implements Gauge<Double> {
+public class KafkaMetricMutableWrapper implements Gauge<Double> {
private org.apache.kafka.common.Metric kafkaMetric;
- public KafkaMetricMuttableWrapper(org.apache.kafka.common.Metric metric) {
+ public KafkaMetricMutableWrapper(org.apache.kafka.common.Metric metric) {
this.kafkaMetric = metric;
}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java
index aa083a20535..f7bed64d556 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka011AvroTableSourceTest extends KafkaAvroTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka011AvroTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
index 451795a5bd5..1a851c1c662 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka011JsonTableSourceTest extends KafkaJsonTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka011JsonTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java
index f4614761d21..61ec9efda5c 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java
@@ -56,7 +56,7 @@ protected String getKafkaVersion() {
}
@Override
- protected KafkaTableSource getExpectedKafkaTableSource(
+ protected KafkaTableSourceBase getExpectedKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -81,7 +81,7 @@ protected KafkaTableSource getExpectedKafkaTableSource(
}
@Override
- protected KafkaTableSink getExpectedKafkaTableSink(
+ protected KafkaTableSinkBase getExpectedKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java
index 189a9fdf46b..b7474e23651 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java
@@ -30,7 +30,7 @@
import java.util.Properties;
/**
- * Kafka 0.8 {@link KafkaTableSink} that serializes data in JSON format.
+ * Kafka 0.8 {@link KafkaTableSinkBase} that serializes data in JSON format.
*
* @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
* with descriptors for schema and format instead. Descriptors allow for
@@ -41,7 +41,7 @@
public class Kafka08JsonTableSink extends KafkaJsonTableSink {
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.8
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.8
* topic with fixed partition assignment.
*
* <p>Each parallel TableSink instance will write its rows to a single Kafka partition.</p>
@@ -62,7 +62,7 @@ public Kafka08JsonTableSink(String topic, Properties properties) {
}
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.8
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.8
* topic with custom partition assignment.
*
* @param topic topic in Kafka to which table is written
@@ -76,7 +76,7 @@ public Kafka08JsonTableSink(String topic, Properties properties, FlinkKafkaParti
}
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.8
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.8
* topic with custom partition assignment.
*
* @param topic topic in Kafka to which table is written
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSink.java b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSink.java
index 146cfc90739..90c42588eaf 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSink.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSink.java
@@ -31,7 +31,7 @@
* Kafka 0.8 table sink for writing data into Kafka.
*/
@Internal
-public class Kafka08TableSink extends KafkaTableSink {
+public class Kafka08TableSink extends KafkaTableSinkBase {
public Kafka08TableSink(
TableSchema schema,
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java
index 97c293e0690..b356abe2ac2 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java
@@ -36,7 +36,7 @@
* Kafka {@link StreamTableSource} for Kafka 0.8.
*/
@Internal
-public class Kafka08TableSource extends KafkaTableSource {
+public class Kafka08TableSource extends KafkaTableSourceBase {
/**
* Creates a Kafka 0.8 {@link StreamTableSource}.
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactory.java b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactory.java
index aeccd4f1ac3..c1d657952cf 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactory.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactory.java
@@ -49,7 +49,7 @@ protected boolean supportsKafkaTimestamps() {
}
@Override
- protected KafkaTableSource createKafkaTableSource(
+ protected KafkaTableSourceBase createKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -73,7 +73,7 @@ protected KafkaTableSource createKafkaTableSource(
}
@Override
- protected KafkaTableSink createKafkaTableSink(
+ protected KafkaTableSinkBase createKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08AvroTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08AvroTableSourceTest.java
index c7b8e8c91d1..bc7b9b6d1ef 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08AvroTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08AvroTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka08AvroTableSourceTest extends KafkaAvroTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka08AvroTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java
index fc46ad4c6ee..aa5fa1609bd 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java
@@ -32,10 +32,10 @@
* drop support for format-specific table sinks.
*/
@Deprecated
-public class Kafka08JsonTableSinkTest extends KafkaTableSinkTestBase {
+public class Kafka08JsonTableSinkTest extends KafkaTableSinkBaseTestBase {
@Override
- protected KafkaTableSink createTableSink(
+ protected KafkaTableSinkBase createTableSink(
String topic,
Properties properties,
FlinkKafkaPartitioner<Row> partitioner) {
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java
index 4ce79ddb8ee..dba8f7dee4e 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka08JsonTableSourceTest extends KafkaJsonTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka08JsonTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactoryTest.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactoryTest.java
index ff633ec0246..ba9b6864f8d 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactoryTest.java
@@ -56,7 +56,7 @@ protected String getKafkaVersion() {
}
@Override
- protected KafkaTableSource getExpectedKafkaTableSource(
+ protected KafkaTableSourceBase getExpectedKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -81,7 +81,7 @@ protected KafkaTableSource getExpectedKafkaTableSource(
}
@Override
- protected KafkaTableSink getExpectedKafkaTableSink(
+ protected KafkaTableSinkBase getExpectedKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSink.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSink.java
index 33634590061..cd27f8371c2 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSink.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSink.java
@@ -30,7 +30,7 @@
import java.util.Properties;
/**
- * Kafka 0.9 {@link KafkaTableSink} that serializes data in JSON format.
+ * Kafka 0.9 {@link KafkaTableSinkBase} that serializes data in JSON format.
*
* @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
* with descriptors for schema and format instead. Descriptors allow for
@@ -41,7 +41,7 @@
public class Kafka09JsonTableSink extends KafkaJsonTableSink {
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.9
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.9
* topic with fixed partition assignment.
*
* <p>Each parallel TableSink instance will write its rows to a single Kafka partition.</p>
@@ -62,7 +62,7 @@ public Kafka09JsonTableSink(String topic, Properties properties) {
}
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.9
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.9
* topic with custom partition assignment.
*
* @param topic topic in Kafka to which table is written
@@ -76,7 +76,7 @@ public Kafka09JsonTableSink(String topic, Properties properties, FlinkKafkaParti
}
/**
- * Creates {@link KafkaTableSink} to write table rows as JSON-encoded records to a Kafka 0.9
+ * Creates {@link KafkaTableSinkBase} to write table rows as JSON-encoded records to a Kafka 0.9
* topic with custom partition assignment.
*
* @param topic topic in Kafka to which table is written
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSink.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSink.java
index 6e38aad1a39..657d75a1c02 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSink.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSink.java
@@ -31,7 +31,7 @@
* Kafka 0.9 table sink for writing data into Kafka.
*/
@Internal
-public class Kafka09TableSink extends KafkaTableSink {
+public class Kafka09TableSink extends KafkaTableSinkBase {
public Kafka09TableSink(
TableSchema schema,
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSource.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSource.java
index 8f9e799ec96..085aa6717af 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSource.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSource.java
@@ -36,7 +36,7 @@
* Kafka {@link StreamTableSource} for Kafka 0.9.
*/
@Internal
-public class Kafka09TableSource extends KafkaTableSource {
+public class Kafka09TableSource extends KafkaTableSourceBase {
/**
* Creates a Kafka 0.9 {@link StreamTableSource}.
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactory.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactory.java
index 19f51508b93..11d56b4491c 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactory.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactory.java
@@ -49,7 +49,7 @@ protected boolean supportsKafkaTimestamps() {
}
@Override
- protected KafkaTableSource createKafkaTableSource(
+ protected KafkaTableSourceBase createKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -73,7 +73,7 @@ protected KafkaTableSource createKafkaTableSource(
}
@Override
- protected KafkaTableSink createKafkaTableSink(
+ protected KafkaTableSinkBase createKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
index dcc67d5b4fb..852d56f86be 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
@@ -202,8 +202,8 @@ protected String getFetcherName() {
return "Kafka 0.9 Fetcher";
}
- protected KafkaConsumerCallBridge createCallBridge() {
- return new KafkaConsumerCallBridge();
+ protected KafkaConsumerCallBridge09 createCallBridge() {
+ return new KafkaConsumerCallBridge09();
}
// ------------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge09.java
similarity index 98%
rename from flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge.java
rename to flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge09.java
index b789633333d..775b7d72f64 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge09.java
@@ -36,7 +36,7 @@
* are compiled against different dependencies.
*/
@Internal
-public class KafkaConsumerCallBridge {
+public class KafkaConsumerCallBridge09 {
public void assignPartitions(KafkaConsumer<?, ?> consumer, List<TopicPartition> topicPartitions) throws Exception {
consumer.assign(topicPartitions);
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java
index 38e8a41d474..7003c3f398d 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java
@@ -81,7 +81,7 @@
private final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue;
/** The indirections on KafkaConsumer methods, for cases where KafkaConsumer compatibility is broken. */
- private final KafkaConsumerCallBridge consumerCallBridge;
+ private final KafkaConsumerCallBridge09 consumerCallBridge;
/** The maximum number of milliseconds to wait for a fetch batch. */
private final long pollTimeout;
@@ -125,7 +125,7 @@ public KafkaConsumerThread(
Handover handover,
Properties kafkaProperties,
ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue,
- KafkaConsumerCallBridge consumerCallBridge,
+ KafkaConsumerCallBridge09 consumerCallBridge,
String threadName,
long pollTimeout,
boolean useMetrics,
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09AvroTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09AvroTableSourceTest.java
index 6f3a56639b3..5f5e80d3347 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09AvroTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09AvroTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka09AvroTableSourceTest extends KafkaAvroTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka09AvroTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java
index 97b5c7d88a2..29cfa93ef2e 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java
@@ -32,10 +32,10 @@
* drop support for format-specific table sinks.
*/
@Deprecated
-public class Kafka09JsonTableSinkTest extends KafkaTableSinkTestBase {
+public class Kafka09JsonTableSinkTest extends KafkaTableSinkBaseTestBase {
@Override
- protected KafkaTableSink createTableSink(
+ protected KafkaTableSinkBase createTableSink(
String topic,
Properties properties,
FlinkKafkaPartitioner<Row> partitioner) {
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java
index bfff7d20102..1a630b1007c 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java
@@ -32,7 +32,7 @@
public class Kafka09JsonTableSourceTest extends KafkaJsonTableSourceTestBase {
@Override
- protected KafkaTableSource.Builder getBuilder() {
+ protected KafkaTableSourceBase.Builder getBuilder() {
return Kafka09JsonTableSource.builder();
}
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactoryTest.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactoryTest.java
index d54c3945949..631a45f974d 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactoryTest.java
@@ -56,7 +56,7 @@ protected String getKafkaVersion() {
}
@Override
- protected KafkaTableSource getExpectedKafkaTableSource(
+ protected KafkaTableSourceBase getExpectedKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -81,7 +81,7 @@ protected KafkaTableSource getExpectedKafkaTableSource(
}
@Override
- protected KafkaTableSink getExpectedKafkaTableSink(
+ protected KafkaTableSinkBase getExpectedKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThreadTest.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThreadTest.java
index 8dbf871ccdb..5da497614fc 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThreadTest.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThreadTest.java
@@ -716,7 +716,7 @@ public TestKafkaConsumerThread(
handover,
new Properties(),
unassignedPartitionsQueue,
- new KafkaConsumerCallBridge(),
+ new KafkaConsumerCallBridge09(),
"test-kafka-consumer-thread",
0,
false,
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java
index 86fd21d15c6..1e9568fd0f9 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java
@@ -44,7 +44,7 @@
*/
@Deprecated
@Internal
-public abstract class KafkaAvroTableSource extends KafkaTableSource {
+public abstract class KafkaAvroTableSource extends KafkaTableSourceBase {
/**
* Creates a generic Kafka Avro {@link StreamTableSource} using a given {@link SpecificRecord}.
@@ -93,7 +93,7 @@ public String explainSource() {
*/
@Deprecated
protected abstract static class Builder<T extends KafkaAvroTableSource, B extends KafkaAvroTableSource.Builder>
- extends KafkaTableSource.Builder<T, B> {
+ extends KafkaTableSourceBase.Builder<T, B> {
private Class<? extends SpecificRecordBase> avroClass;
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
index 231edddb311..d84eb891044 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
@@ -28,13 +28,13 @@
import java.util.Properties;
/**
- * Base class for {@link KafkaTableSink} that serializes data in JSON format.
+ * Base class for {@link KafkaTableSinkBase} that serializes data in JSON format.
*
* @deprecated Use table descriptors instead of implementation-specific classes.
*/
@Deprecated
@Internal
-public abstract class KafkaJsonTableSink extends KafkaTableSink {
+public abstract class KafkaJsonTableSink extends KafkaTableSinkBase {
/**
* Creates KafkaJsonTableSink.
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
index 70b286bc296..a9db97938d6 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
@@ -43,7 +43,7 @@
*/
@Deprecated
@Internal
-public abstract class KafkaJsonTableSource extends KafkaTableSource {
+public abstract class KafkaJsonTableSource extends KafkaTableSourceBase {
/**
* Creates a generic Kafka JSON {@link StreamTableSource}.
@@ -102,7 +102,7 @@ protected void setFailOnMissingField(boolean failOnMissingField) {
*/
@Deprecated
protected abstract static class Builder<T extends KafkaJsonTableSource, B extends KafkaJsonTableSource.Builder>
- extends KafkaTableSource.Builder<T, B> {
+ extends KafkaTableSourceBase.Builder<T, B> {
private TableSchema jsonSchema;
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBase.java
similarity index 94%
rename from flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
rename to flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBase.java
index a85d536eac9..acd10cc8855 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBase.java
@@ -43,7 +43,7 @@
* override {@link #createKafkaProducer(String, Properties, SerializationSchema, Optional)}}.
*/
@Internal
-public abstract class KafkaTableSink implements AppendStreamTableSink<Row> {
+public abstract class KafkaTableSinkBase implements AppendStreamTableSink<Row> {
// TODO make all attributes final and mandatory once we drop support for format-specific table sinks
@@ -66,7 +66,7 @@
protected String[] fieldNames;
protected TypeInformation[] fieldTypes;
- protected KafkaTableSink(
+ protected KafkaTableSinkBase(
TableSchema schema,
String topic,
Properties properties,
@@ -81,7 +81,7 @@ protected KafkaTableSink(
}
/**
- * Creates KafkaTableSink.
+ * Creates KafkaTableSinkBase.
*
* @param topic Kafka topic to write to.
* @param properties Properties for the Kafka producer.
@@ -89,7 +89,7 @@ protected KafkaTableSink(
* @deprecated Use table descriptors instead of implementation-specific classes.
*/
@Deprecated
- public KafkaTableSink(
+ public KafkaTableSinkBase(
String topic,
Properties properties,
FlinkKafkaPartitioner<Row> partitioner) {
@@ -133,7 +133,7 @@ public KafkaTableSink(
* @return Deep copy of this sink
*/
@Deprecated
- protected KafkaTableSink createCopy() {
+ protected KafkaTableSinkBase createCopy() {
throw new UnsupportedOperationException("This method only exists for backwards compatibility.");
}
@@ -164,14 +164,14 @@ public void emitDataStream(DataStream<Row> dataStream) {
}
@Override
- public KafkaTableSink configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
+ public KafkaTableSinkBase configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
if (schema.isPresent()) {
// a fixed schema is defined so reconfiguration is not supported
throw new UnsupportedOperationException("Reconfiguration of this sink is not supported.");
}
// legacy code
- KafkaTableSink copy = createCopy();
+ KafkaTableSinkBase copy = createCopy();
copy.fieldNames = Preconditions.checkNotNull(fieldNames, "fieldNames");
copy.fieldTypes = Preconditions.checkNotNull(fieldTypes, "fieldTypes");
Preconditions.checkArgument(fieldNames.length == fieldTypes.length,
@@ -191,7 +191,7 @@ public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) {
return false;
}
- KafkaTableSink that = (KafkaTableSink) o;
+ KafkaTableSinkBase that = (KafkaTableSinkBase) o;
return Objects.equals(schema, that.schema) &&
Objects.equals(topic, that.topic) &&
Objects.equals(properties, that.properties) &&
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBase.java
similarity index 97%
rename from flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
rename to flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBase.java
index 474c22f704f..04aa8c1ea8c 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBase.java
@@ -57,7 +57,7 @@
* override {@link #createKafkaConsumer(String, Properties, DeserializationSchema)}}.
*/
@Internal
-public abstract class KafkaTableSource implements
+public abstract class KafkaTableSourceBase implements
StreamTableSource<Row>,
DefinedProctimeAttribute,
DefinedRowtimeAttributes,
@@ -110,7 +110,7 @@
* @param specificStartupOffsets Specific startup offsets; only relevant when startup
* mode is {@link StartupMode#SPECIFIC_OFFSETS}.
*/
- protected KafkaTableSource(
+ protected KafkaTableSourceBase(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -141,7 +141,7 @@ protected KafkaTableSource(
* @param properties Properties for the Kafka consumer.
* @param deserializationSchema Deserialization schema for decoding records from Kafka.
*/
- protected KafkaTableSource(
+ protected KafkaTableSourceBase(
TableSchema schema,
String topic,
Properties properties,
@@ -225,10 +225,10 @@ public boolean equals(Object o) {
}
// TODO force classes to be equal once we drop support for format-specific table sources
// if (o == null || getClass() != o.getClass()) {
- if (o == null || !(o instanceof KafkaTableSource)) {
+ if (o == null || !(o instanceof KafkaTableSourceBase)) {
return false;
}
- final KafkaTableSource that = (KafkaTableSource) o;
+ final KafkaTableSourceBase that = (KafkaTableSourceBase) o;
return Objects.equals(schema, that.schema) &&
Objects.equals(proctimeAttribute, that.proctimeAttribute) &&
Objects.equals(rowtimeAttributeDescriptors, that.rowtimeAttributeDescriptors) &&
@@ -398,18 +398,18 @@ protected void setFieldMapping(Map<String, String> fieldMapping) {
DeserializationSchema<Row> deserializationSchema);
/**
- * Abstract builder for a {@link KafkaTableSource} to be extended by builders of subclasses of
- * KafkaTableSource.
+ * Abstract builder for a {@link KafkaTableSourceBase} to be extended by builders of subclasses of
+ * KafkaTableSourceBase.
*
- * @param <T> Type of the KafkaTableSource produced by the builder.
- * @param <B> Type of the KafkaTableSource.Builder subclass.
+ * @param <T> Type of the KafkaTableSourceBase produced by the builder.
+ * @param <B> Type of the KafkaTableSourceBase.Builder subclass.
* @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
* with descriptors for schema and format instead. Descriptors allow for
* implementation-agnostic definition of tables. See also
* {@link org.apache.flink.table.api.TableEnvironment#connect(ConnectorDescriptor)}.
*/
@Deprecated
- protected abstract static class Builder<T extends KafkaTableSource, B extends KafkaTableSource.Builder> {
+ protected abstract static class Builder<T extends KafkaTableSourceBase, B extends KafkaTableSourceBase.Builder> {
private String topic;
@@ -682,11 +682,11 @@ protected void configureTableSource(T tableSource) {
protected abstract B builder();
/**
- * Builds the configured {@link KafkaTableSource}.
- * @return The configured {@link KafkaTableSource}.
+ * Builds the configured {@link KafkaTableSourceBase}.
+ * @return The configured {@link KafkaTableSourceBase}.
* @deprecated Use table descriptors instead of implementation-specific builders.
*/
@Deprecated
- protected abstract KafkaTableSource build();
+ protected abstract KafkaTableSourceBase build();
}
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java
index 5634331adbb..431f946ae4a 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java
@@ -83,7 +83,7 @@
import static org.apache.flink.table.descriptors.StreamTableDescriptorValidator.UPDATE_MODE_VALUE_APPEND;
/**
- * Factory for creating configured instances of {@link KafkaTableSource}.
+ * Factory for creating configured instances of {@link KafkaTableSourceBase}.
*/
public abstract class KafkaTableSourceSinkFactoryBase implements
StreamTableSourceFactory<Row>,
@@ -213,7 +213,7 @@
* @param specificStartupOffsets Specific startup offsets; only relevant when startup
* mode is {@link StartupMode#SPECIFIC_OFFSETS}.
*/
- protected abstract KafkaTableSource createKafkaTableSource(
+ protected abstract KafkaTableSourceBase createKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -232,7 +232,7 @@ protected abstract KafkaTableSource createKafkaTableSource(
* @param properties Properties for the Kafka consumer.
* @param partitioner Partitioner to select Kafka partition for each item.
*/
- protected abstract KafkaTableSink createKafkaTableSink(
+ protected abstract KafkaTableSinkBase createKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java
index cad37f8f8cd..c7ddb7f39ee 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java
@@ -36,6 +36,7 @@
public static final String CONNECTOR_VERSION_VALUE_09 = "0.9";
public static final String CONNECTOR_VERSION_VALUE_010 = "0.10";
public static final String CONNECTOR_VERSION_VALUE_011 = "0.11";
+ public static final String CONNECTOR_VERSION_VALUE_20 = "2.0";
public static final String CONNECTOR_TOPIC = "connector.topic";
public static final String CONNECTOR_STARTUP_MODE = "connector.startup-mode";
public static final String CONNECTOR_STARTUP_MODE_VALUE_EARLIEST = "earliest-offset";
@@ -73,7 +74,8 @@ private void validateVersion(DescriptorProperties properties) {
CONNECTOR_VERSION_VALUE_08,
CONNECTOR_VERSION_VALUE_09,
CONNECTOR_VERSION_VALUE_010,
- CONNECTOR_VERSION_VALUE_011);
+ CONNECTOR_VERSION_VALUE_011,
+ CONNECTOR_VERSION_VALUE_20);
properties.validateEnumValues(CONNECTOR_VERSION(), false, versions);
properties.validateString(CONNECTOR_TOPIC, false, 1, Integer.MAX_VALUE);
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java
index 140ce21d942..089a1182518 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java
@@ -43,7 +43,7 @@
public abstract class KafkaAvroTableSourceTestBase extends KafkaTableSourceBuilderTestBase {
@Override
- protected void configureBuilder(KafkaTableSource.Builder builder) {
+ protected void configureBuilder(KafkaTableSourceBase.Builder builder) {
super.configureBuilder(builder);
((KafkaAvroTableSource.Builder) builder).forAvroRecordClass(SchemaRecord.class);
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
index 7d88f0d94ea..4ce8e62ca32 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
@@ -78,12 +78,6 @@
import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
-import kafka.consumer.Consumer;
-import kafka.consumer.ConsumerConfig;
-import kafka.consumer.ConsumerIterator;
-import kafka.consumer.KafkaStream;
-import kafka.javaapi.consumer.ConsumerConnector;
-import kafka.message.MessageAndMetadata;
import kafka.server.KafkaServer;
import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.kafka.clients.producer.ProducerConfig;
@@ -184,7 +178,10 @@ public void runFailOnNoBrokerTest() throws Exception {
stream.print();
see.execute("No broker test");
} catch (JobExecutionException jee) {
- if (kafkaServer.getVersion().equals("0.9") || kafkaServer.getVersion().equals("0.10") || kafkaServer.getVersion().equals("0.11")) {
+ if (kafkaServer.getVersion().equals("0.9") ||
+ kafkaServer.getVersion().equals("0.10") ||
+ kafkaServer.getVersion().equals("0.11") ||
+ kafkaServer.getVersion().equals("2.0")) {
assertTrue(jee.getCause() instanceof TimeoutException);
TimeoutException te = (TimeoutException) jee.getCause();
@@ -2115,41 +2112,6 @@ public void run() {
// Debugging utilities
// ------------------------------------------------------------------------
- /**
- * Read topic to list, only using Kafka code.
- */
- private static List<MessageAndMetadata<byte[], byte[]>> readTopicToList(String topicName, ConsumerConfig config, final int stopAfter) {
- ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(config);
- // we request only one stream per consumer instance. Kafka will make sure that each consumer group
- // will see each message only once.
- Map<String, Integer> topicCountMap = Collections.singletonMap(topicName, 1);
- Map<String, List<KafkaStream<byte[], byte[]>>> streams = consumerConnector.createMessageStreams(topicCountMap);
- if (streams.size() != 1) {
- throw new RuntimeException("Expected only one message stream but got " + streams.size());
- }
- List<KafkaStream<byte[], byte[]>> kafkaStreams = streams.get(topicName);
- if (kafkaStreams == null) {
- throw new RuntimeException("Requested stream not available. Available streams: " + streams.toString());
- }
- if (kafkaStreams.size() != 1) {
- throw new RuntimeException("Requested 1 stream from Kafka, bot got " + kafkaStreams.size() + " streams");
- }
- LOG.info("Opening Consumer instance for topic '{}' on group '{}'", topicName, config.groupId());
- ConsumerIterator<byte[], byte[]> iteratorToRead = kafkaStreams.get(0).iterator();
-
- List<MessageAndMetadata<byte[], byte[]>> result = new ArrayList<>();
- int read = 0;
- while (iteratorToRead.hasNext()) {
- read++;
- result.add(iteratorToRead.next());
- if (read == stopAfter) {
- LOG.info("Read " + read + " elements");
- return result;
- }
- }
- return result;
- }
-
private static class BrokerKillingMapper<T> extends RichMapFunction<T, T>
implements ListCheckpointed<Integer>, CheckpointListener {
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java
index 51c0e7bbde2..afb95175d52 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java
@@ -114,7 +114,7 @@ private void testTableSource(FormatDescriptor format) {
specificOffsets.put(new KafkaTopicPartition(TOPIC, 0), 100L);
specificOffsets.put(new KafkaTopicPartition(TOPIC, 1), 123L);
- final KafkaTableSource builderSource = builder()
+ final KafkaTableSourceBase builderSource = builder()
.forJsonSchema(TableSchema.fromTypeInfo(JsonRowSchemaConverter.convert(JSON_SCHEMA)))
.failOnMissingField(true)
.withTableToJsonMapping(tableJsonMapping)
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBaseTestBase.java
similarity index 90%
rename from flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkTestBase.java
rename to flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBaseTestBase.java
index b4bb89dc048..dbbb10f51bf 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBaseTestBase.java
@@ -50,7 +50,7 @@
* drop support for format-specific table sinks.
*/
@Deprecated
-public abstract class KafkaTableSinkTestBase {
+public abstract class KafkaTableSinkBaseTestBase {
private static final String TOPIC = "testTopic";
private static final String[] FIELD_NAMES = new String[] {"field1", "field2"};
@@ -64,7 +64,7 @@ public void testKafkaTableSink() {
DataStream dataStream = mock(DataStream.class);
when(dataStream.addSink(any(SinkFunction.class))).thenReturn(mock(DataStreamSink.class));
- KafkaTableSink kafkaTableSink = spy(createTableSink());
+ KafkaTableSinkBase kafkaTableSink = spy(createTableSink());
kafkaTableSink.emitDataStream(dataStream);
// verify correct producer class
@@ -80,8 +80,8 @@ public void testKafkaTableSink() {
@Test
public void testConfiguration() {
- KafkaTableSink kafkaTableSink = createTableSink();
- KafkaTableSink newKafkaTableSink = kafkaTableSink.configure(FIELD_NAMES, FIELD_TYPES);
+ KafkaTableSinkBase kafkaTableSink = createTableSink();
+ KafkaTableSinkBase newKafkaTableSink = kafkaTableSink.configure(FIELD_NAMES, FIELD_TYPES);
assertNotSame(kafkaTableSink, newKafkaTableSink);
assertArrayEquals(FIELD_NAMES, newKafkaTableSink.getFieldNames());
@@ -89,7 +89,7 @@ public void testConfiguration() {
assertEquals(new RowTypeInfo(FIELD_TYPES), newKafkaTableSink.getOutputType());
}
- protected abstract KafkaTableSink createTableSink(
+ protected abstract KafkaTableSinkBase createTableSink(
String topic,
Properties properties,
FlinkKafkaPartitioner<Row> partitioner);
@@ -98,8 +98,8 @@ protected abstract KafkaTableSink createTableSink(
protected abstract Class<? extends FlinkKafkaProducerBase> getProducerClass();
- private KafkaTableSink createTableSink() {
- KafkaTableSink sink = createTableSink(TOPIC, PROPERTIES, PARTITIONER);
+ private KafkaTableSinkBase createTableSink() {
+ KafkaTableSinkBase sink = createTableSink(TOPIC, PROPERTIES, PARTITIONER);
return sink.configure(FIELD_NAMES, FIELD_TYPES);
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBuilderTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBuilderTestBase.java
index b9ae47b3f54..16dc714fb44 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBuilderTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBuilderTestBase.java
@@ -70,11 +70,11 @@
@Test
@SuppressWarnings("unchecked")
public void testKafkaConsumer() {
- KafkaTableSource.Builder b = getBuilder();
+ KafkaTableSourceBase.Builder b = getBuilder();
configureBuilder(b);
// assert that correct
- KafkaTableSource observed = spy(b.build());
+ KafkaTableSourceBase observed = spy(b.build());
StreamExecutionEnvironment env = mock(StreamExecutionEnvironment.class);
when(env.addSource(any(SourceFunction.class))).thenReturn(mock(DataStreamSource.class));
observed.getDataStream(env);
@@ -89,10 +89,10 @@ public void testKafkaConsumer() {
@Test
public void testTableSchema() {
- KafkaTableSource.Builder b = getBuilder();
+ KafkaTableSourceBase.Builder b = getBuilder();
configureBuilder(b);
- KafkaTableSource source = b.build();
+ KafkaTableSourceBase source = b.build();
// check table schema
TableSchema schema = source.getTableSchema();
@@ -113,10 +113,10 @@ public void testTableSchema() {
@Test
public void testNoTimeAttributes() {
- KafkaTableSource.Builder b = getBuilder();
+ KafkaTableSourceBase.Builder b = getBuilder();
configureBuilder(b);
- KafkaTableSource source = b.build();
+ KafkaTableSourceBase source = b.build();
// assert no proctime
assertNull(source.getProctimeAttribute());
@@ -127,11 +127,11 @@ public void testNoTimeAttributes() {
@Test
public void testProctimeAttribute() {
- KafkaTableSource.Builder b = getBuilder();
+ KafkaTableSourceBase.Builder b = getBuilder();
configureBuilder(b);
b.withProctimeAttribute("time1");
- KafkaTableSource source = b.build();
+ KafkaTableSourceBase source = b.build();
// assert correct proctime field
assertEquals(source.getProctimeAttribute(), "time1");
@@ -143,11 +143,11 @@ public void testProctimeAttribute() {
@Test
public void testRowtimeAttribute() {
- KafkaTableSource.Builder b = getBuilder();
+ KafkaTableSourceBase.Builder b = getBuilder();
configureBuilder(b);
b.withRowtimeAttribute("time2", new ExistingField("time2"), new AscendingTimestamps());
- KafkaTableSource source = b.build();
+ KafkaTableSourceBase source = b.build();
// assert no proctime
assertNull(source.getProctimeAttribute());
@@ -168,13 +168,13 @@ public void testRowtimeAttribute() {
@Test
public void testRowtimeAttribute2() {
- KafkaTableSource.Builder b = getBuilder();
+ KafkaTableSourceBase.Builder b = getBuilder();
configureBuilder(b);
try {
b.withKafkaTimestampAsRowtimeAttribute("time2", new AscendingTimestamps());
- KafkaTableSource source = b.build();
+ KafkaTableSourceBase source = b.build();
// assert no proctime
assertNull(source.getProctimeAttribute());
@@ -201,11 +201,11 @@ public void testRowtimeAttribute2() {
@Test
@SuppressWarnings("unchecked")
public void testConsumerOffsets() {
- KafkaTableSource.Builder b = getBuilder();
+ KafkaTableSourceBase.Builder b = getBuilder();
configureBuilder(b);
// test the default behavior
- KafkaTableSource source = spy(b.build());
+ KafkaTableSourceBase source = spy(b.build());
when(source.createKafkaConsumer(TOPIC, PROPS, null))
.thenReturn(mock(getFlinkKafkaConsumer()));
@@ -241,13 +241,13 @@ public void testConsumerOffsets() {
verify(source.getKafkaConsumer(TOPIC, PROPS, null)).setStartFromSpecificOffsets(any(Map.class));
}
- protected abstract KafkaTableSource.Builder getBuilder();
+ protected abstract KafkaTableSourceBase.Builder getBuilder();
protected abstract Class<DeserializationSchema<Row>> getDeserializationSchema();
protected abstract Class<FlinkKafkaConsumerBase<Row>> getFlinkKafkaConsumer();
- protected void configureBuilder(KafkaTableSource.Builder builder) {
+ protected void configureBuilder(KafkaTableSourceBase.Builder builder) {
builder
.forTopic(TOPIC)
.withKafkaProperties(PROPS)
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java
index 5e9144c1e57..135e7183adf 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java
@@ -133,7 +133,7 @@ public void testTableSource() {
.toRowType()
);
- final KafkaTableSource expected = getExpectedKafkaTableSource(
+ final KafkaTableSourceBase expected = getExpectedKafkaTableSource(
schema,
Optional.of(PROC_TIME),
rowtimeAttributeDescriptors,
@@ -172,7 +172,7 @@ public void testTableSource() {
assertEquals(expected, actualSource);
// test Kafka consumer
- final KafkaTableSource actualKafkaSource = (KafkaTableSource) actualSource;
+ final KafkaTableSourceBase actualKafkaSource = (KafkaTableSourceBase) actualSource;
final StreamExecutionEnvironmentMock mock = new StreamExecutionEnvironmentMock();
actualKafkaSource.getDataStream(mock);
assertTrue(getExpectedFlinkKafkaConsumer().isAssignableFrom(mock.sourceFunction.getClass()));
@@ -191,7 +191,7 @@ public void testTableSink() {
.field(EVENT_TIME, Types.SQL_TIMESTAMP())
.build();
- final KafkaTableSink expected = getExpectedKafkaTableSink(
+ final KafkaTableSinkBase expected = getExpectedKafkaTableSink(
schema,
TOPIC,
KAFKA_PROPERTIES,
@@ -222,7 +222,7 @@ public void testTableSink() {
assertEquals(expected, actualSink);
// test Kafka producer
- final KafkaTableSink actualKafkaSink = (KafkaTableSink) actualSink;
+ final KafkaTableSinkBase actualKafkaSink = (KafkaTableSinkBase) actualSink;
final DataStreamMock streamMock = new DataStreamMock(new StreamExecutionEnvironmentMock(), schema.toRowType());
actualKafkaSink.emitDataStream(streamMock);
assertTrue(getExpectedFlinkKafkaProducer().isAssignableFrom(streamMock.sinkFunction.getClass()));
@@ -286,7 +286,7 @@ public void setChainingStrategy(ChainingStrategy strategy) {
protected abstract Class<?> getExpectedFlinkKafkaProducer();
- protected abstract KafkaTableSource getExpectedKafkaTableSource(
+ protected abstract KafkaTableSourceBase getExpectedKafkaTableSource(
TableSchema schema,
Optional<String> proctimeAttribute,
List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
@@ -297,7 +297,7 @@ protected abstract KafkaTableSource getExpectedKafkaTableSource(
StartupMode startupMode,
Map<KafkaTopicPartition, Long> specificStartupOffsets);
- protected abstract KafkaTableSink getExpectedKafkaTableSink(
+ protected abstract KafkaTableSinkBase getExpectedKafkaTableSink(
TableSchema schema,
String topic,
Properties properties,
diff --git a/flink-connectors/flink-connector-kafka/pom.xml b/flink-connectors/flink-connector-kafka/pom.xml
new file mode 100644
index 00000000000..109535aa4db
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/pom.xml
@@ -0,0 +1,318 @@
+<?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/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>flink-connectors</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>1.7-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+ <name>flink-connector-kafka</name>
+
+ <packaging>jar</packaging>
+
+ <properties>
+ <kafka.version>2.0.0</kafka.version>
+ </properties>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-base_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.binary.version}</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <!-- streaming-java dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Add Kafka 2.0.x as a dependency -->
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</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.flink</groupId>
+ <artifactId>flink-avro</artifactId>
+ <version>${project.version}</version>
+ <!-- Projects depending on this project, won't depend on flink-avro. -->
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <!-- Projects depending on this project, won't depend on flink-json. -->
+ <optional>true</optional>
+ </dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-avro</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-base_${scala.binary.version}</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>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-base_${scala.binary.version}</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>
+
+ <dependency>
+ <!-- include 2.0 server for tests -->
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${scala.binary.version}</artifactId>
+ <version>${kafka.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </exclusion>
+ </exclusions>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-tests_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics-jmx</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <profiles>
+ <!-- Create SQL Client uber jars by default -->
+ <profile>
+ <id>sql-jars</id>
+ <activation>
+ <property>
+ <name>!skipSqlJars</name>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadedArtifactAttached>true</shadedArtifactAttached>
+ <shadedClassifierName>sql-jar</shadedClassifierName>
+ <artifactSet>
+ <includes combine.children="append">
+ <include>org.apache.kafka:*</include>
+ <include>org.apache.flink:flink-connector-kafka-base_${scala.binary.version}</include>
+ </includes>
+ </artifactSet>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>kafka/kafka-version.properties</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <relocations>
+ <relocation>
+ <pattern>org.apache.kafka</pattern>
+ <shadedPattern>org.apache.flink.kafka.shaded.org.apache.kafka</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+ <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 -Xmx2048m -Dlog4j.configuration=${log4j.configuration} -Dmvn.forkNumber=${surefire.forkNumber} -XX:-UseGCOverheadLimit</argLine>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java
new file mode 100644
index 00000000000..ddbe4217b2a
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java
@@ -0,0 +1,321 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.metrics.MetricGroup;
+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.config.OffsetCommitMode;
+import org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher;
+import org.apache.flink.streaming.connectors.kafka.internal.KafkaPartitionDiscoverer;
+import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
+import org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicsDescriptor;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper;
+import org.apache.flink.util.PropertiesUtil;
+import org.apache.flink.util.SerializedValue;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.regex.Pattern;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.PropertiesUtil.getBoolean;
+import static org.apache.flink.util.PropertiesUtil.getLong;
+
+/**
+ * The Flink Kafka Consumer is a streaming data source that pulls a parallel data stream from
+ * Apache Kafka. The consumer can run in multiple parallel instances, each of which will pull
+ * data from one or more Kafka partitions.
+ *
+ * <p>The Flink Kafka Consumer participates in checkpointing and guarantees that no data is lost
+ * during a failure, and that the computation processes elements "exactly once".
+ * (Note: These guarantees naturally assume that Kafka itself does not loose any data.)</p>
+ *
+ * <p>Please note that Flink snapshots the offsets internally as part of its distributed checkpoints. The offsets
+ * committed to Kafka 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>
+ */
+@PublicEvolving
+public class FlinkKafkaConsumer<T> extends FlinkKafkaConsumerBase<T> {
+
+ /** Configuration key to change the polling timeout. **/
+ public static final String KEY_POLL_TIMEOUT = "flink.poll-timeout";
+
+ /** 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. **/
+ 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 */
+ protected final long pollTimeout;
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a new Kafka streaming source consumer.
+ *
+ * @param topic The name of the topic that should be consumed.
+ * @param valueDeserializer The de-/serializer used to convert between Kafka's byte messages and Flink's objects.
+ * @param props
+ */
+ public FlinkKafkaConsumer(String topic, DeserializationSchema<T> valueDeserializer, Properties props) {
+ this(Collections.singletonList(topic), valueDeserializer, props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer.
+ *
+ * <p>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
+ */
+ public FlinkKafkaConsumer(String topic, KeyedDeserializationSchema<T> deserializer, Properties props) {
+ this(Collections.singletonList(topic), deserializer, props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer.
+ *
+ * <p>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
+ */
+ public FlinkKafkaConsumer(List<String> topics, DeserializationSchema<T> deserializer, Properties props) {
+ this(topics, new KeyedDeserializationSchemaWrapper<>(deserializer), props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer.
+ *
+ * <p>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
+ */
+ public FlinkKafkaConsumer(List<String> topics, KeyedDeserializationSchema<T> deserializer, Properties props) {
+ this(topics, null, deserializer, props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer. Use this constructor to
+ * subscribe to multiple topics based on a regular expression pattern.
+ *
+ * <p>If partition discovery is enabled (by setting a non-negative value for
+ * {@link FlinkKafkaConsumer#KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS} in the properties), topics
+ * with names matching the pattern will also be subscribed to as they are created on the fly.
+ *
+ * @param subscriptionPattern The regular expression for a pattern of topic names to subscribe to.
+ * @param valueDeserializer The de-/serializer used to convert between Kafka's byte messages and Flink's objects.
+ * @param props
+ */
+ public FlinkKafkaConsumer(Pattern subscriptionPattern, DeserializationSchema<T> valueDeserializer, Properties props) {
+ this(null, subscriptionPattern, new KeyedDeserializationSchemaWrapper<>(valueDeserializer), props);
+ }
+
+ /**
+ * Creates a new Kafka streaming source consumer. Use this constructor to
+ * subscribe to multiple topics based on a regular expression pattern.
+ *
+ * <p>If partition discovery is enabled (by setting a non-negative value for
+ * {@link FlinkKafkaConsumer#KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS} in the properties), topics
+ * with names matching the pattern will also be subscribed to as they are created on the fly.
+ *
+ * <p>This constructor allows passing a {@see KeyedDeserializationSchema} for reading key/value
+ * pairs, offsets, and topic names from Kafka.
+ *
+ * @param subscriptionPattern The regular expression for a pattern of topic names to subscribe to.
+ * @param deserializer The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects.
+ * @param props
+ */
+ public FlinkKafkaConsumer(Pattern subscriptionPattern, KeyedDeserializationSchema<T> deserializer, Properties props) {
+ this(null, subscriptionPattern, deserializer, props);
+ }
+
+ private FlinkKafkaConsumer(
+ List<String> topics,
+ Pattern subscriptionPattern,
+ KeyedDeserializationSchema<T> deserializer,
+ Properties props) {
+
+ super(
+ topics,
+ subscriptionPattern,
+ deserializer,
+ getLong(
+ checkNotNull(props, "props"),
+ KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, PARTITION_DISCOVERY_DISABLED),
+ !getBoolean(props, KEY_DISABLE_METRICS, false));
+
+ this.properties = 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);
+ }
+ }
+
+ @Override
+ protected AbstractFetcher<T, ?> createFetcher(
+ SourceContext<T> sourceContext,
+ Map<KafkaTopicPartition, Long> assignedPartitionsWithInitialOffsets,
+ SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
+ SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
+ StreamingRuntimeContext runtimeContext,
+ OffsetCommitMode offsetCommitMode,
+ MetricGroup consumerMetricGroup,
+ boolean useMetrics) throws Exception {
+
+ // make sure that auto commit is disabled when our offset commit mode is ON_CHECKPOINTS;
+ // this overwrites whatever setting the user configured in the properties
+ if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS || offsetCommitMode == OffsetCommitMode.DISABLED) {
+ properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+ }
+
+ return new KafkaFetcher<>(
+ sourceContext,
+ assignedPartitionsWithInitialOffsets,
+ watermarksPeriodic,
+ watermarksPunctuated,
+ runtimeContext.getProcessingTimeService(),
+ runtimeContext.getExecutionConfig().getAutoWatermarkInterval(),
+ runtimeContext.getUserCodeClassLoader(),
+ runtimeContext.getTaskNameWithSubtasks(),
+ deserializer,
+ properties,
+ pollTimeout,
+ runtimeContext.getMetricGroup(),
+ consumerMetricGroup,
+ useMetrics);
+ }
+
+ @Override
+ protected AbstractPartitionDiscoverer createPartitionDiscoverer(
+ KafkaTopicsDescriptor topicsDescriptor,
+ int indexOfThisSubtask,
+ int numParallelSubtasks) {
+
+ return new KafkaPartitionDiscoverer(topicsDescriptor, indexOfThisSubtask, numParallelSubtasks, properties);
+ }
+
+ // ------------------------------------------------------------------------
+ // Timestamp-based startup
+ // ------------------------------------------------------------------------
+
+ @Override
+ public FlinkKafkaConsumerBase<T> setStartFromTimestamp(long startupOffsetsTimestamp) {
+ // the purpose of this override is just to publicly expose the method for Kafka 0.10+;
+ // the base class doesn't publicly expose it since not all Kafka versions support the functionality
+ return super.setStartFromTimestamp(startupOffsetsTimestamp);
+ }
+
+ @Override
+ protected Map<KafkaTopicPartition, Long> fetchOffsetsWithTimestamp(
+ Collection<KafkaTopicPartition> partitions,
+ long timestamp) {
+
+ Map<TopicPartition, Long> partitionOffsetsRequest = new HashMap<>(partitions.size());
+ for (KafkaTopicPartition partition : partitions) {
+ partitionOffsetsRequest.put(
+ new TopicPartition(partition.getTopic(), partition.getPartition()),
+ timestamp);
+ }
+
+ // use a short-lived consumer to fetch the offsets;
+ // this is ok because this is a one-time operation that happens only on startup
+ KafkaConsumer<?, ?> consumer = new KafkaConsumer(properties);
+
+ Map<KafkaTopicPartition, Long> result = new HashMap<>(partitions.size());
+ for (Map.Entry<TopicPartition, OffsetAndTimestamp> partitionToOffset :
+ consumer.offsetsForTimes(partitionOffsetsRequest).entrySet()) {
+
+ result.put(
+ new KafkaTopicPartition(partitionToOffset.getKey().topic(), partitionToOffset.getKey().partition()),
+ (partitionToOffset.getValue() == null) ? null : partitionToOffset.getValue().offset());
+ }
+
+ consumer.close();
+ return result;
+ }
+
+ @Override
+ protected boolean getIsAutoCommitEnabled() {
+ return getBoolean(properties, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true) &&
+ PropertiesUtil.getLong(properties, ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000) > 0;
+ }
+
+ /**
+ * 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.getName();
+
+ 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);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaErrorCode.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaErrorCode.java
new file mode 100644
index 00000000000..492098eb2f9
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaErrorCode.java
@@ -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.
+ */
+
+package org.apache.flink.streaming.connectors.kafka;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Error codes used in {@link FlinkKafkaException}.
+ */
+@PublicEvolving
+public enum FlinkKafkaErrorCode {
+ PRODUCERS_POOL_EMPTY,
+ EXTERNAL_ERROR
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaException.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaException.java
new file mode 100644
index 00000000000..30f7d531fcd
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaException.java
@@ -0,0 +1,46 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.util.FlinkException;
+
+/**
+ * Exception used by {@link FlinkKafkaProducer} and {@link FlinkKafkaConsumer}.
+ */
+@PublicEvolving
+public class FlinkKafkaException extends FlinkException {
+
+ private static final long serialVersionUID = 920269130311214200L;
+
+ private final FlinkKafkaErrorCode errorCode;
+
+ public FlinkKafkaException(FlinkKafkaErrorCode errorCode, String message) {
+ super(message);
+ this.errorCode = errorCode;
+ }
+
+ public FlinkKafkaException(FlinkKafkaErrorCode errorCode, String message, Throwable cause) {
+ super(message, cause);
+ this.errorCode = errorCode;
+ }
+
+ public FlinkKafkaErrorCode getErrorCode() {
+ return errorCode;
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
new file mode 100644
index 00000000000..0ac2f906bde
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -0,0 +1,1352 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaInternalProducer;
+import org.apache.flink.streaming.connectors.kafka.internal.TransactionalIdsGenerator;
+import org.apache.flink.streaming.connectors.kafka.internal.metrics.KafkaMetricMutableWrapper;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaDelegatePartitioner;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.NetUtils;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.errors.InvalidTxnStateException;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Flink Sink to produce data into a Kafka topic. By default producer
+ * will use {@link FlinkKafkaProducer.Semantic#AT_LEAST_ONCE} semantic.
+ * Before using {@link FlinkKafkaProducer.Semantic#EXACTLY_ONCE} please refer to Flink's
+ * Kafka connector documentation.
+ */
+@PublicEvolving
+public class FlinkKafkaProducer<IN>
+ extends TwoPhaseCommitSinkFunction<IN, FlinkKafkaProducer.KafkaTransactionState, FlinkKafkaProducer.KafkaTransactionContext> {
+
+ /**
+ * Semantics that can be chosen.
+ * <li>{@link #EXACTLY_ONCE}</li>
+ * <li>{@link #AT_LEAST_ONCE}</li>
+ * <li>{@link #NONE}</li>
+ */
+ public enum Semantic {
+
+ /**
+ * Semantic.EXACTLY_ONCE the Flink producer will write all messages in a Kafka transaction that will be
+ * committed to Kafka on a checkpoint.
+ *
+ * <p>In this mode {@link FlinkKafkaProducer} sets up a pool of {@link FlinkKafkaInternalProducer}. Between each
+ * checkpoint a Kafka transaction is created, which is committed on
+ * {@link FlinkKafkaProducer#notifyCheckpointComplete(long)}. If checkpoint complete notifications are
+ * running late, {@link FlinkKafkaProducer} can run out of {@link FlinkKafkaInternalProducer}s in the pool. In that
+ * case any subsequent {@link FlinkKafkaProducer#snapshotState(FunctionSnapshotContext)} requests will fail
+ * and {@link FlinkKafkaProducer} will keep using the {@link FlinkKafkaInternalProducer}
+ * from the previous checkpoint.
+ * To decrease the chance of failing checkpoints there are three options:
+ * <li>decrease number of max concurrent checkpoints</li>
+ * <li>make checkpoints more reliable (so that they complete faster)</li>
+ * <li>increase the delay between checkpoints</li>
+ * <li>increase the size of {@link FlinkKafkaInternalProducer}s pool</li>
+ */
+ EXACTLY_ONCE,
+
+ /**
+ * Semantic.AT_LEAST_ONCE the Flink producer will wait for all outstanding messages in the Kafka buffers
+ * to be acknowledged by the Kafka producer on a checkpoint.
+ */
+ AT_LEAST_ONCE,
+
+ /**
+ * Semantic.NONE means that nothing will be guaranteed. Messages can be lost and/or duplicated in case
+ * of failure.
+ */
+ NONE
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaProducer.class);
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This coefficient determines what is the safe scale down factor.
+ *
+ * <p>If the Flink application previously failed before first checkpoint completed or we are starting new batch
+ * of {@link FlinkKafkaProducer} from scratch without clean shutdown of the previous one,
+ * {@link FlinkKafkaProducer} doesn't know what was the set of previously used Kafka's transactionalId's. In
+ * that case, it will try to play safe and abort all of the possible transactionalIds from the range of:
+ * {@code [0, getNumberOfParallelSubtasks() * kafkaProducersPoolSize * SAFE_SCALE_DOWN_FACTOR) }
+ *
+ * <p>The range of available to use transactional ids is:
+ * {@code [0, getNumberOfParallelSubtasks() * kafkaProducersPoolSize) }
+ *
+ * <p>This means that if we decrease {@code getNumberOfParallelSubtasks()} by a factor larger then
+ * {@code SAFE_SCALE_DOWN_FACTOR} we can have a left some lingering transaction.
+ */
+ public static final int SAFE_SCALE_DOWN_FACTOR = 5;
+
+ /**
+ * Default number of KafkaProducers in the pool. See {@link FlinkKafkaProducer.Semantic#EXACTLY_ONCE}.
+ */
+ public static final int DEFAULT_KAFKA_PRODUCERS_POOL_SIZE = 5;
+
+ /**
+ * Default value for kafka transaction timeout.
+ */
+ public static final Time DEFAULT_KAFKA_TRANSACTION_TIMEOUT = Time.hours(1);
+
+ /**
+ * Configuration key for disabling the metrics reporting.
+ */
+ public static final String KEY_DISABLE_METRICS = "flink.disable-metrics";
+
+ /**
+ * Descriptor of the transactional IDs list.
+ */
+ private static final ListStateDescriptor<FlinkKafkaProducer.NextTransactionalIdHint> NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR =
+ new ListStateDescriptor<>("next-transactional-id-hint", TypeInformation.of(FlinkKafkaProducer.NextTransactionalIdHint.class));
+
+ /**
+ * State for nextTransactionalIdHint.
+ */
+ private transient ListState<FlinkKafkaProducer.NextTransactionalIdHint> nextTransactionalIdHintState;
+
+ /**
+ * Generator for Transactional IDs.
+ */
+ private transient TransactionalIdsGenerator transactionalIdsGenerator;
+
+ /**
+ * Hint for picking next transactional id.
+ */
+ private transient FlinkKafkaProducer.NextTransactionalIdHint nextTransactionalIdHint;
+
+ /**
+ * User defined properties for the Producer.
+ */
+ protected final Properties producerConfig;
+
+ /**
+ * The name of the default topic this producer is writing data to.
+ */
+ private final String defaultTopicId;
+
+ /**
+ * (Serializable) SerializationSchema for turning objects used with Flink into.
+ * byte[] for Kafka.
+ */
+ private final KeyedSerializationSchema<IN> schema;
+
+ /**
+ * User-provided partitioner for assigning an object to a Kafka partition for each topic.
+ */
+ private final FlinkKafkaPartitioner<IN> flinkKafkaPartitioner;
+
+ /**
+ * Partitions of each topic.
+ */
+ private final Map<String, int[]> topicPartitionsMap;
+
+ /**
+ * Max number of producers in the pool. If all producers are in use, snapshoting state will throw an exception.
+ */
+ private final int kafkaProducersPoolSize;
+
+ /**
+ * Pool of available transactional ids.
+ */
+ private final BlockingDeque<String> availableTransactionalIds = new LinkedBlockingDeque<>();
+
+ /**
+ * Flag controlling whether we are writing the Flink record's timestamp into Kafka.
+ */
+ private boolean writeTimestampToKafka = false;
+
+ /**
+ * Flag indicating whether to accept failures (and log them), or to fail on failures.
+ */
+ private boolean logFailuresOnly;
+
+ /**
+ * Semantic chosen for this instance.
+ */
+ protected FlinkKafkaProducer.Semantic semantic;
+
+ // -------------------------------- Runtime fields ------------------------------------------
+
+ /** The callback than handles error propagation or logging callbacks. */
+ @Nullable
+ private transient Callback callback;
+
+ /** Errors encountered in the async producer are stored here. */
+ @Nullable
+ private transient volatile Exception asyncException;
+
+ /** Number of unacknowledged records. */
+ private final AtomicLong pendingRecords = new AtomicLong();
+
+ /** Cache of metrics to replace already registered metrics instead of overwriting existing ones. */
+ private final Map<String, KafkaMetricMutableWrapper> previouslyCreatedMetrics = new HashMap<>();
+
+ /**
+ * 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 FlinkKafkaProducer(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) {
+ this(
+ topicId,
+ new KeyedSerializationSchemaWrapper<>(serializationSchema),
+ getPropertiesFromBrokerList(brokerList),
+ Optional.of(new FlinkFixedPartitioner<IN>()));
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * <p>Using this constructor, the default {@link FlinkFixedPartitioner} will be used as
+ * the partitioner. This default partitioner maps each sink subtask to a single Kafka
+ * partition (i.e. all records received by a sink subtask will end up in the same
+ * Kafka partition).
+ *
+ * <p>To use a custom partitioner, please use
+ * {@link #FlinkKafkaProducer(String, SerializationSchema, Properties, Optional)} instead.
+ *
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined key-less serialization schema.
+ * @param producerConfig
+ * Properties with the producer configuration.
+ */
+ public FlinkKafkaProducer(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) {
+ this(
+ topicId,
+ new KeyedSerializationSchemaWrapper<>(serializationSchema),
+ producerConfig,
+ Optional.of(new FlinkFixedPartitioner<IN>()));
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to
+ * the topic. It accepts a key-less {@link SerializationSchema} and possibly a custom {@link FlinkKafkaPartitioner}.
+ *
+ * <p>Since a key-less {@link SerializationSchema} is used, all records sent to Kafka will not have an
+ * attached key. Therefore, if a partitioner is also not provided, records will be distributed to Kafka
+ * partitions in a round-robin fashion.
+ *
+ * @param topicId The topic to write data to
+ * @param serializationSchema A key-less 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.
+ * If a partitioner is not provided, records will be distributed to Kafka partitions
+ * in a round-robin fashion.
+ */
+ public FlinkKafkaProducer(
+ String topicId,
+ SerializationSchema<IN> serializationSchema,
+ Properties producerConfig,
+ Optional<FlinkKafkaPartitioner<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.
+ *
+ * <p>Using this constructor, the default {@link FlinkFixedPartitioner} will be used as
+ * the partitioner. This default partitioner maps each sink subtask to a single Kafka
+ * partition (i.e. all records received by a sink subtask will end up in the same
+ * Kafka partition).
+ *
+ * <p>To use a custom partitioner, please use
+ * {@link #FlinkKafkaProducer(String, KeyedSerializationSchema, Properties, Optional)} instead.
+ *
+ * @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 FlinkKafkaProducer(String brokerList, String topicId, KeyedSerializationSchema<IN> serializationSchema) {
+ this(
+ topicId,
+ serializationSchema,
+ getPropertiesFromBrokerList(brokerList),
+ Optional.of(new FlinkFixedPartitioner<IN>()));
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * <p>Using this constructor, the default {@link FlinkFixedPartitioner} will be used as
+ * the partitioner. This default partitioner maps each sink subtask to a single Kafka
+ * partition (i.e. all records received by a sink subtask will end up in the same
+ * Kafka partition).
+ *
+ * <p>To use a custom partitioner, please use
+ * {@link #FlinkKafkaProducer(String, KeyedSerializationSchema, Properties, Optional)} instead.
+ *
+ * @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 FlinkKafkaProducer(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig) {
+ this(
+ topicId,
+ serializationSchema,
+ producerConfig,
+ Optional.of(new FlinkFixedPartitioner<IN>()));
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
+ * the topic.
+ *
+ * <p>Using this constructor, the default {@link FlinkFixedPartitioner} will be used as
+ * the partitioner. This default partitioner maps each sink subtask to a single Kafka
+ * partition (i.e. all records received by a sink subtask will end up in the same
+ * Kafka partition).
+ *
+ * <p>To use a custom partitioner, please use
+ * {@link #FlinkKafkaProducer(String, KeyedSerializationSchema, Properties, Optional, FlinkKafkaProducer.Semantic, int)} instead.
+ *
+ * @param topicId
+ * ID of the Kafka topic.
+ * @param serializationSchema
+ * User defined serialization schema supporting key/value messages
+ * @param producerConfig
+ * Properties with the producer configuration.
+ * @param semantic
+ * Defines semantic that will be used by this producer (see {@link FlinkKafkaProducer.Semantic}).
+ */
+ public FlinkKafkaProducer(
+ String topicId,
+ KeyedSerializationSchema<IN> serializationSchema,
+ Properties producerConfig,
+ FlinkKafkaProducer.Semantic semantic) {
+ this(topicId,
+ serializationSchema,
+ producerConfig,
+ Optional.of(new FlinkFixedPartitioner<IN>()),
+ semantic,
+ DEFAULT_KAFKA_PRODUCERS_POOL_SIZE);
+ }
+
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to
+ * the topic. It accepts a keyed {@link KeyedSerializationSchema} and possibly a custom {@link FlinkKafkaPartitioner}.
+ *
+ * <p>If a partitioner is not provided, written records will be partitioned by the attached key of each
+ * record (as determined by {@link KeyedSerializationSchema#serializeKey(Object)}). If written records do not
+ * have a key (i.e., {@link KeyedSerializationSchema#serializeKey(Object)} returns {@code null}), they
+ * will be distributed to Kafka partitions in a round-robin fashion.
+ *
+ * @param defaultTopicId The default 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.
+ * If a partitioner is not provided, records will be partitioned by the key of each record
+ * (determined by {@link KeyedSerializationSchema#serializeKey(Object)}). If the keys
+ * are {@code null}, then records will be distributed to Kafka partitions in a
+ * round-robin fashion.
+ */
+ public FlinkKafkaProducer(
+ String defaultTopicId,
+ KeyedSerializationSchema<IN> serializationSchema,
+ Properties producerConfig,
+ Optional<FlinkKafkaPartitioner<IN>> customPartitioner) {
+ this(
+ defaultTopicId,
+ serializationSchema,
+ producerConfig,
+ customPartitioner,
+ FlinkKafkaProducer.Semantic.AT_LEAST_ONCE,
+ DEFAULT_KAFKA_PRODUCERS_POOL_SIZE);
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to
+ * the topic. It accepts a keyed {@link KeyedSerializationSchema} and possibly a custom {@link FlinkKafkaPartitioner}.
+ *
+ * <p>If a partitioner is not provided, written records will be partitioned by the attached key of each
+ * record (as determined by {@link KeyedSerializationSchema#serializeKey(Object)}). If written records do not
+ * have a key (i.e., {@link KeyedSerializationSchema#serializeKey(Object)} returns {@code null}), they
+ * will be distributed to Kafka partitions in a round-robin fashion.
+ *
+ * @param defaultTopicId The default 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.
+ * If a partitioner is not provided, records will be partitioned by the key of each record
+ * (determined by {@link KeyedSerializationSchema#serializeKey(Object)}). If the keys
+ * are {@code null}, then records will be distributed to Kafka partitions in a
+ * round-robin fashion.
+ * @param semantic Defines semantic that will be used by this producer (see {@link FlinkKafkaProducer.Semantic}).
+ * @param kafkaProducersPoolSize Overwrite default KafkaProducers pool size (see {@link FlinkKafkaProducer.Semantic#EXACTLY_ONCE}).
+ */
+ public FlinkKafkaProducer(
+ String defaultTopicId,
+ KeyedSerializationSchema<IN> serializationSchema,
+ Properties producerConfig,
+ Optional<FlinkKafkaPartitioner<IN>> customPartitioner,
+ FlinkKafkaProducer.Semantic semantic,
+ int kafkaProducersPoolSize) {
+ super(new FlinkKafkaProducer.TransactionStateSerializer(), new FlinkKafkaProducer.ContextStateSerializer());
+
+ this.defaultTopicId = checkNotNull(defaultTopicId, "defaultTopicId is null");
+ this.schema = checkNotNull(serializationSchema, "serializationSchema is null");
+ this.producerConfig = checkNotNull(producerConfig, "producerConfig is null");
+ this.flinkKafkaPartitioner = checkNotNull(customPartitioner, "customPartitioner is null").orElse(null);
+ this.semantic = checkNotNull(semantic, "semantic is null");
+ this.kafkaProducersPoolSize = kafkaProducersPoolSize;
+ checkState(kafkaProducersPoolSize > 0, "kafkaProducersPoolSize must be non empty");
+
+ ClosureCleaner.clean(this.flinkKafkaPartitioner, true);
+ ClosureCleaner.ensureSerializable(serializationSchema);
+
+ // set the producer configuration properties for kafka record key value serializers.
+ if (!producerConfig.containsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) {
+ this.producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+ } else {
+ LOG.warn("Overwriting the '{}' is not recommended", ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);
+ }
+
+ if (!producerConfig.containsKey(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)) {
+ this.producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+ } else {
+ LOG.warn("Overwriting the '{}' is not recommended", ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);
+ }
+
+ // eagerly ensure that bootstrap servers are set.
+ if (!this.producerConfig.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
+ throw new IllegalArgumentException(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " must be supplied in the producer config properties.");
+ }
+
+ if (!producerConfig.containsKey(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG)) {
+ long timeout = DEFAULT_KAFKA_TRANSACTION_TIMEOUT.toMilliseconds();
+ checkState(timeout < Integer.MAX_VALUE && timeout > 0, "timeout does not fit into 32 bit integer");
+ this.producerConfig.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, (int) timeout);
+ LOG.warn("Property [{}] not specified. Setting it to {}", ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, DEFAULT_KAFKA_TRANSACTION_TIMEOUT);
+ }
+
+ // Enable transactionTimeoutWarnings to avoid silent data loss
+ // See KAFKA-6119 (affects versions 0.11.0.0 and 0.11.0.1):
+ // The KafkaProducer may not throw an exception if the transaction failed to commit
+ if (semantic == FlinkKafkaProducer.Semantic.EXACTLY_ONCE) {
+ final Object object = this.producerConfig.get(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG);
+ final long transactionTimeout;
+ if (object instanceof String && StringUtils.isNumeric((String) object)) {
+ transactionTimeout = Long.parseLong((String) object);
+ } else if (object instanceof Number) {
+ transactionTimeout = ((Number) object).longValue();
+ } else {
+ throw new IllegalArgumentException(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG
+ + " must be numeric, was " + object);
+ }
+ super.setTransactionTimeout(transactionTimeout);
+ super.enableTransactionTimeoutWarnings(0.8);
+ }
+
+ this.topicPartitionsMap = new HashMap<>();
+ }
+
+ // ---------------------------------- Properties --------------------------
+
+ /**
+ * 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.writeTimestampToKafka = writeTimestampToKafka;
+ }
+
+ /**
+ * 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.logFailuresOnly = logFailuresOnly;
+ }
+
+ /**
+ * Disables the propagation of exceptions thrown when committing presumably timed out Kafka
+ * transactions during recovery of the job. If a Kafka transaction is timed out, a commit will
+ * never be successful. Hence, use this feature to avoid recovery loops of the Job. Exceptions
+ * will still be logged to inform the user that data loss might have occurred.
+ *
+ * <p>Note that we use {@link System#currentTimeMillis()} to track the age of a transaction.
+ * Moreover, only exceptions thrown during the recovery are caught, i.e., the producer will
+ * attempt at least one commit of the transaction before giving up.</p>
+ */
+ @Override
+ public FlinkKafkaProducer<IN> ignoreFailuresAfterTransactionTimeout() {
+ super.ignoreFailuresAfterTransactionTimeout();
+ return this;
+ }
+
+ // ----------------------------------- Utilities --------------------------
+
+ /**
+ * Initializes the connection to Kafka.
+ */
+ @Override
+ public void open(Configuration configuration) throws Exception {
+ if (logFailuresOnly) {
+ callback = new Callback() {
+ @Override
+ public void onCompletion(RecordMetadata metadata, Exception e) {
+ if (e != null) {
+ LOG.error("Error while sending record to Kafka: " + e.getMessage(), e);
+ }
+ acknowledgeMessage();
+ }
+ };
+ }
+ else {
+ callback = new Callback() {
+ @Override
+ public void onCompletion(RecordMetadata metadata, Exception exception) {
+ if (exception != null && asyncException == null) {
+ asyncException = exception;
+ }
+ acknowledgeMessage();
+ }
+ };
+ }
+
+ super.open(configuration);
+ }
+
+ @Override
+ public void invoke(FlinkKafkaProducer.KafkaTransactionState transaction, IN next, Context context) throws FlinkKafkaException {
+ checkErroneous();
+
+ byte[] serializedKey = schema.serializeKey(next);
+ byte[] serializedValue = schema.serializeValue(next);
+ String targetTopic = schema.getTargetTopic(next);
+ if (targetTopic == null) {
+ targetTopic = defaultTopicId;
+ }
+
+ Long timestamp = null;
+ if (this.writeTimestampToKafka) {
+ timestamp = context.timestamp();
+ }
+
+ ProducerRecord<byte[], byte[]> record;
+ int[] partitions = topicPartitionsMap.get(targetTopic);
+ if (null == partitions) {
+ partitions = getPartitionsByTopic(targetTopic, transaction.producer);
+ topicPartitionsMap.put(targetTopic, partitions);
+ }
+ if (flinkKafkaPartitioner != null) {
+ record = new ProducerRecord<>(
+ targetTopic,
+ flinkKafkaPartitioner.partition(next, serializedKey, serializedValue, targetTopic, partitions),
+ timestamp,
+ serializedKey,
+ serializedValue);
+ } else {
+ record = new ProducerRecord<>(targetTopic, null, timestamp, serializedKey, serializedValue);
+ }
+ pendingRecords.incrementAndGet();
+ transaction.producer.send(record, callback);
+ }
+
+ @Override
+ public void close() throws FlinkKafkaException {
+ final FlinkKafkaProducer.KafkaTransactionState currentTransaction = currentTransaction();
+ if (currentTransaction != null) {
+ // to avoid exceptions on aborting transactions with some pending records
+ flush(currentTransaction);
+
+ // normal abort for AT_LEAST_ONCE and NONE do not clean up resources because of producer reusing, thus
+ // we need to close it manually
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ break;
+ case AT_LEAST_ONCE:
+ case NONE:
+ currentTransaction.producer.close();
+ break;
+ }
+ }
+ try {
+ super.close();
+ }
+ catch (Exception e) {
+ asyncException = ExceptionUtils.firstOrSuppressed(e, asyncException);
+ }
+ // make sure we propagate pending errors
+ checkErroneous();
+ }
+
+ // ------------------- Logic for handling checkpoint flushing -------------------------- //
+
+ @Override
+ protected FlinkKafkaProducer.KafkaTransactionState beginTransaction() throws FlinkKafkaException {
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ FlinkKafkaInternalProducer<byte[], byte[]> producer = createTransactionalProducer();
+ producer.beginTransaction();
+ return new FlinkKafkaProducer.KafkaTransactionState(producer.getTransactionalId(), producer);
+ case AT_LEAST_ONCE:
+ case NONE:
+ // Do not create new producer on each beginTransaction() if it is not necessary
+ final FlinkKafkaProducer.KafkaTransactionState currentTransaction = currentTransaction();
+ if (currentTransaction != null && currentTransaction.producer != null) {
+ return new FlinkKafkaProducer.KafkaTransactionState(currentTransaction.producer);
+ }
+ return new FlinkKafkaProducer.KafkaTransactionState(initProducer(true));
+ default:
+ throw new UnsupportedOperationException("Not implemented semantic");
+ }
+ }
+
+ @Override
+ protected void preCommit(FlinkKafkaProducer.KafkaTransactionState transaction) throws FlinkKafkaException {
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ case AT_LEAST_ONCE:
+ flush(transaction);
+ break;
+ case NONE:
+ break;
+ default:
+ throw new UnsupportedOperationException("Not implemented semantic");
+ }
+ checkErroneous();
+ }
+
+ @Override
+ protected void commit(FlinkKafkaProducer.KafkaTransactionState transaction) {
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ transaction.producer.commitTransaction();
+ recycleTransactionalProducer(transaction.producer);
+ break;
+ case AT_LEAST_ONCE:
+ case NONE:
+ break;
+ default:
+ throw new UnsupportedOperationException("Not implemented semantic");
+ }
+ }
+
+ @Override
+ protected void recoverAndCommit(FlinkKafkaProducer.KafkaTransactionState transaction) {
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ try (FlinkKafkaInternalProducer<byte[], byte[]> producer =
+ initTransactionalProducer(transaction.transactionalId, false)) {
+ producer.resumeTransaction(transaction.producerId, transaction.epoch);
+ producer.commitTransaction();
+ }
+ catch (InvalidTxnStateException | ProducerFencedException ex) {
+ // That means we have committed this transaction before.
+ LOG.warn("Encountered error {} while recovering transaction {}. " +
+ "Presumably this transaction has been already committed before",
+ ex,
+ transaction);
+ }
+ break;
+ case AT_LEAST_ONCE:
+ case NONE:
+ break;
+ default:
+ throw new UnsupportedOperationException("Not implemented semantic");
+ }
+ }
+
+ @Override
+ protected void abort(FlinkKafkaProducer.KafkaTransactionState transaction) {
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ transaction.producer.abortTransaction();
+ recycleTransactionalProducer(transaction.producer);
+ break;
+ case AT_LEAST_ONCE:
+ case NONE:
+ break;
+ default:
+ throw new UnsupportedOperationException("Not implemented semantic");
+ }
+ }
+
+ @Override
+ protected void recoverAndAbort(FlinkKafkaProducer.KafkaTransactionState transaction) {
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ try (FlinkKafkaInternalProducer<byte[], byte[]> producer =
+ initTransactionalProducer(transaction.transactionalId, false)) {
+ producer.initTransactions();
+ }
+ break;
+ case AT_LEAST_ONCE:
+ case NONE:
+ break;
+ default:
+ throw new UnsupportedOperationException("Not implemented semantic");
+ }
+ }
+
+ private void acknowledgeMessage() {
+ pendingRecords.decrementAndGet();
+ }
+
+ /**
+ * Flush pending records.
+ * @param transaction
+ */
+ private void flush(FlinkKafkaProducer.KafkaTransactionState transaction) throws FlinkKafkaException {
+ if (transaction.producer != null) {
+ transaction.producer.flush();
+ }
+ long pendingRecordsCount = pendingRecords.get();
+ if (pendingRecordsCount != 0) {
+ throw new IllegalStateException("Pending record count must be zero at this point: " + pendingRecordsCount);
+ }
+
+ // if the flushed requests has errors, we should propagate it also and fail the checkpoint
+ checkErroneous();
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+
+ nextTransactionalIdHintState.clear();
+ // To avoid duplication only first subtask keeps track of next transactional id hint. Otherwise all of the
+ // subtasks would write exactly same information.
+ if (getRuntimeContext().getIndexOfThisSubtask() == 0 && semantic == FlinkKafkaProducer.Semantic.EXACTLY_ONCE) {
+ checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be set for EXACTLY_ONCE");
+ long nextFreeTransactionalId = nextTransactionalIdHint.nextFreeTransactionalId;
+
+ // If we scaled up, some (unknown) subtask must have created new transactional ids from scratch. In that
+ // case we adjust nextFreeTransactionalId by the range of transactionalIds that could be used for this
+ // scaling up.
+ if (getRuntimeContext().getNumberOfParallelSubtasks() > nextTransactionalIdHint.lastParallelism) {
+ nextFreeTransactionalId += getRuntimeContext().getNumberOfParallelSubtasks() * kafkaProducersPoolSize;
+ }
+
+ nextTransactionalIdHintState.add(new FlinkKafkaProducer.NextTransactionalIdHint(
+ getRuntimeContext().getNumberOfParallelSubtasks(),
+ nextFreeTransactionalId));
+ }
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ if (semantic != FlinkKafkaProducer.Semantic.NONE && !((StreamingRuntimeContext) this.getRuntimeContext()).isCheckpointingEnabled()) {
+ LOG.warn("Using {} semantic, but checkpointing is not enabled. Switching to {} semantic.", semantic, FlinkKafkaProducer.Semantic.NONE);
+ semantic = FlinkKafkaProducer.Semantic.NONE;
+ }
+
+ nextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(
+ NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
+ transactionalIdsGenerator = new TransactionalIdsGenerator(
+ getRuntimeContext().getTaskName() + "-" + ((StreamingRuntimeContext) getRuntimeContext()).getOperatorUniqueID(),
+ getRuntimeContext().getIndexOfThisSubtask(),
+ getRuntimeContext().getNumberOfParallelSubtasks(),
+ kafkaProducersPoolSize,
+ SAFE_SCALE_DOWN_FACTOR);
+
+ if (semantic != FlinkKafkaProducer.Semantic.EXACTLY_ONCE) {
+ nextTransactionalIdHint = null;
+ } else {
+ ArrayList<FlinkKafkaProducer.NextTransactionalIdHint> transactionalIdHints = Lists.newArrayList(nextTransactionalIdHintState.get());
+ if (transactionalIdHints.size() > 1) {
+ throw new IllegalStateException(
+ "There should be at most one next transactional id hint written by the first subtask");
+ } else if (transactionalIdHints.size() == 0) {
+ nextTransactionalIdHint = new FlinkKafkaProducer.NextTransactionalIdHint(0, 0);
+
+ // this means that this is either:
+ // (1) the first execution of this application
+ // (2) previous execution has failed before first checkpoint completed
+ //
+ // in case of (2) we have to abort all previous transactions
+ abortTransactions(transactionalIdsGenerator.generateIdsToAbort());
+ } else {
+ nextTransactionalIdHint = transactionalIdHints.get(0);
+ }
+ }
+
+ super.initializeState(context);
+ }
+
+ @Override
+ protected Optional<FlinkKafkaProducer.KafkaTransactionContext> initializeUserContext() {
+ if (semantic != FlinkKafkaProducer.Semantic.EXACTLY_ONCE) {
+ return Optional.empty();
+ }
+
+ Set<String> transactionalIds = generateNewTransactionalIds();
+ resetAvailableTransactionalIdsPool(transactionalIds);
+ return Optional.of(new FlinkKafkaProducer.KafkaTransactionContext(transactionalIds));
+ }
+
+ private Set<String> generateNewTransactionalIds() {
+ checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be present for EXACTLY_ONCE");
+
+ Set<String> transactionalIds = transactionalIdsGenerator.generateIdsToUse(nextTransactionalIdHint.nextFreeTransactionalId);
+ LOG.info("Generated new transactionalIds {}", transactionalIds);
+ return transactionalIds;
+ }
+
+ @Override
+ protected void finishRecoveringContext() {
+ cleanUpUserContext();
+ resetAvailableTransactionalIdsPool(getUserContext().get().transactionalIds);
+ LOG.info("Recovered transactionalIds {}", getUserContext().get().transactionalIds);
+ }
+
+ protected FlinkKafkaInternalProducer createProducer() {
+ return new FlinkKafkaInternalProducer<>(this.producerConfig);
+ }
+
+ /**
+ * After initialization make sure that all previous transactions from the current user context have been completed.
+ */
+ private void cleanUpUserContext() {
+ if (!getUserContext().isPresent()) {
+ return;
+ }
+ abortTransactions(getUserContext().get().transactionalIds);
+ }
+
+ private void resetAvailableTransactionalIdsPool(Collection<String> transactionalIds) {
+ availableTransactionalIds.clear();
+ for (String transactionalId : transactionalIds) {
+ availableTransactionalIds.add(transactionalId);
+ }
+ }
+
+ // ----------------------------------- Utilities --------------------------
+
+ private void abortTransactions(Set<String> transactionalIds) {
+ for (String transactionalId : transactionalIds) {
+ try (FlinkKafkaInternalProducer<byte[], byte[]> kafkaProducer =
+ initTransactionalProducer(transactionalId, false)) {
+ // it suffice to call initTransactions - this will abort any lingering transactions
+ kafkaProducer.initTransactions();
+ }
+ }
+ }
+
+ int getTransactionCoordinatorId() {
+ final FlinkKafkaProducer.KafkaTransactionState currentTransaction = currentTransaction();
+ if (currentTransaction == null || currentTransaction.producer == null) {
+ throw new IllegalArgumentException();
+ }
+ return currentTransaction.producer.getTransactionCoordinatorId();
+ }
+
+ /**
+ * For each checkpoint we create new {@link FlinkKafkaInternalProducer} so that new transactions will not clash
+ * with transactions created during previous checkpoints ({@code producer.initTransactions()} assures that we
+ * obtain new producerId and epoch counters).
+ */
+ private FlinkKafkaInternalProducer<byte[], byte[]> createTransactionalProducer() throws FlinkKafkaException {
+ String transactionalId = availableTransactionalIds.poll();
+ if (transactionalId == null) {
+ throw new FlinkKafkaException(
+ FlinkKafkaErrorCode.PRODUCERS_POOL_EMPTY,
+ "Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints.");
+ }
+ FlinkKafkaInternalProducer<byte[], byte[]> producer = initTransactionalProducer(transactionalId, true);
+ producer.initTransactions();
+ return producer;
+ }
+
+ private void recycleTransactionalProducer(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+ availableTransactionalIds.add(producer.getTransactionalId());
+ producer.close();
+ }
+
+ private FlinkKafkaInternalProducer<byte[], byte[]> initTransactionalProducer(String transactionalId, boolean registerMetrics) {
+ producerConfig.put("transactional.id", transactionalId);
+ return initProducer(registerMetrics);
+ }
+
+ private FlinkKafkaInternalProducer<byte[], byte[]> initProducer(boolean registerMetrics) {
+ FlinkKafkaInternalProducer<byte[], byte[]> producer = createProducer();
+
+ RuntimeContext ctx = getRuntimeContext();
+
+ if (flinkKafkaPartitioner != null) {
+ if (flinkKafkaPartitioner instanceof FlinkKafkaDelegatePartitioner) {
+ ((FlinkKafkaDelegatePartitioner) flinkKafkaPartitioner).setPartitions(
+ getPartitionsByTopic(this.defaultTopicId, producer));
+ }
+ flinkKafkaPartitioner.open(ctx.getIndexOfThisSubtask(), ctx.getNumberOfParallelSubtasks());
+ }
+
+ LOG.info("Starting FlinkKafkaInternalProducer ({}/{}) to produce into default topic {}",
+ ctx.getIndexOfThisSubtask() + 1, ctx.getNumberOfParallelSubtasks(), defaultTopicId);
+
+ // register Kafka metrics to Flink accumulators
+ if (registerMetrics && !Boolean.parseBoolean(producerConfig.getProperty(KEY_DISABLE_METRICS, "false"))) {
+ Map<MetricName, ? extends Metric> metrics = producer.metrics();
+
+ if (metrics == null) {
+ // MapR's Kafka implementation returns null here.
+ LOG.info("Producer implementation does not support metrics");
+ } else {
+ final MetricGroup kafkaMetricGroup = getRuntimeContext().getMetricGroup().addGroup("KafkaProducer");
+ for (Map.Entry<MetricName, ? extends Metric> entry: metrics.entrySet()) {
+ String name = entry.getKey().name();
+ Metric metric = entry.getValue();
+
+ KafkaMetricMutableWrapper wrapper = previouslyCreatedMetrics.get(name);
+ if (wrapper != null) {
+ wrapper.setKafkaMetric(metric);
+ } else {
+ // TODO: somehow merge metrics from all active producers?
+ wrapper = new KafkaMetricMutableWrapper(metric);
+ previouslyCreatedMetrics.put(name, wrapper);
+ kafkaMetricGroup.gauge(name, wrapper);
+ }
+ }
+ }
+ }
+ return producer;
+ }
+
+ private void checkErroneous() throws FlinkKafkaException {
+ Exception e = asyncException;
+ if (e != null) {
+ // prevent double throwing
+ asyncException = null;
+ throw new FlinkKafkaException(
+ FlinkKafkaErrorCode.EXTERNAL_ERROR,
+ "Failed to send data to Kafka: " + e.getMessage(),
+ e);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+ }
+
+ private static Properties getPropertiesFromBrokerList(String brokerList) {
+ String[] elements = brokerList.split(",");
+
+ // validate the broker addresses
+ for (String broker: elements) {
+ NetUtils.getCorrectHostnamePort(broker);
+ }
+
+ Properties props = new Properties();
+ props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+ return props;
+ }
+
+ private static int[] getPartitionsByTopic(String topic, Producer<byte[], byte[]> producer) {
+ // the fetched list is immutable, so we're creating a mutable copy in order to sort it
+ List<PartitionInfo> partitionsList = new ArrayList<>(producer.partitionsFor(topic));
+
+ // sort the partitions by partition id to make sure the fetched partition list is the same across subtasks
+ Collections.sort(partitionsList, new Comparator<PartitionInfo>() {
+ @Override
+ public int compare(PartitionInfo o1, PartitionInfo o2) {
+ return Integer.compare(o1.partition(), o2.partition());
+ }
+ });
+
+ int[] partitions = new int[partitionsList.size()];
+ for (int i = 0; i < partitions.length; i++) {
+ partitions[i] = partitionsList.get(i).partition();
+ }
+
+ return partitions;
+ }
+
+ /**
+ * State for handling transactions.
+ */
+ @VisibleForTesting
+ @Internal
+ static class KafkaTransactionState {
+
+ private final transient FlinkKafkaInternalProducer<byte[], byte[]> producer;
+
+ @Nullable
+ final String transactionalId;
+
+ final long producerId;
+
+ final short epoch;
+
+ KafkaTransactionState(String transactionalId, FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+ this(transactionalId, producer.getProducerId(), producer.getEpoch(), producer);
+ }
+
+ KafkaTransactionState(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+ this(null, -1, (short) -1, producer);
+ }
+
+ KafkaTransactionState(
+ String transactionalId,
+ long producerId,
+ short epoch,
+ FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+ this.transactionalId = transactionalId;
+ this.producerId = producerId;
+ this.epoch = epoch;
+ this.producer = producer;
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "%s [transactionalId=%s, producerId=%s, epoch=%s]",
+ this.getClass().getSimpleName(),
+ transactionalId,
+ producerId,
+ epoch);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ FlinkKafkaProducer.KafkaTransactionState that = (FlinkKafkaProducer.KafkaTransactionState) o;
+
+ if (producerId != that.producerId) {
+ return false;
+ }
+ if (epoch != that.epoch) {
+ return false;
+ }
+ return transactionalId != null ? transactionalId.equals(that.transactionalId) : that.transactionalId == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = transactionalId != null ? transactionalId.hashCode() : 0;
+ result = 31 * result + (int) (producerId ^ (producerId >>> 32));
+ result = 31 * result + (int) epoch;
+ return result;
+ }
+ }
+
+ /**
+ * Context associated to this instance of the {@link FlinkKafkaProducer}. User for keeping track of the
+ * transactionalIds.
+ */
+ @VisibleForTesting
+ @Internal
+ public static class KafkaTransactionContext {
+ final Set<String> transactionalIds;
+
+ KafkaTransactionContext(Set<String> transactionalIds) {
+ checkNotNull(transactionalIds);
+ this.transactionalIds = transactionalIds;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ FlinkKafkaProducer.KafkaTransactionContext that = (FlinkKafkaProducer.KafkaTransactionContext) o;
+
+ return transactionalIds.equals(that.transactionalIds);
+ }
+
+ @Override
+ public int hashCode() {
+ return transactionalIds.hashCode();
+ }
+ }
+
+ /**
+ * {@link org.apache.flink.api.common.typeutils.TypeSerializer} for
+ * {@link FlinkKafkaProducer.KafkaTransactionState}.
+ */
+ @VisibleForTesting
+ @Internal
+ public static class TransactionStateSerializer extends TypeSerializerSingleton<FlinkKafkaProducer.KafkaTransactionState> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionState createInstance() {
+ return null;
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionState copy(FlinkKafkaProducer.KafkaTransactionState from) {
+ return from;
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionState copy(
+ FlinkKafkaProducer.KafkaTransactionState from,
+ FlinkKafkaProducer.KafkaTransactionState reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(
+ FlinkKafkaProducer.KafkaTransactionState record,
+ DataOutputView target) throws IOException {
+ if (record.transactionalId == null) {
+ target.writeBoolean(false);
+ } else {
+ target.writeBoolean(true);
+ target.writeUTF(record.transactionalId);
+ }
+ target.writeLong(record.producerId);
+ target.writeShort(record.epoch);
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionState deserialize(DataInputView source) throws IOException {
+ String transactionalId = null;
+ if (source.readBoolean()) {
+ transactionalId = source.readUTF();
+ }
+ long producerId = source.readLong();
+ short epoch = source.readShort();
+ return new FlinkKafkaProducer.KafkaTransactionState(transactionalId, producerId, epoch, null);
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionState deserialize(
+ FlinkKafkaProducer.KafkaTransactionState reuse,
+ DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(
+ DataInputView source, DataOutputView target) throws IOException {
+ boolean hasTransactionalId = source.readBoolean();
+ target.writeBoolean(hasTransactionalId);
+ if (hasTransactionalId) {
+ target.writeUTF(source.readUTF());
+ }
+ target.writeLong(source.readLong());
+ target.writeShort(source.readShort());
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof FlinkKafkaProducer.TransactionStateSerializer;
+ }
+ }
+
+ /**
+ * {@link org.apache.flink.api.common.typeutils.TypeSerializer} for
+ * {@link FlinkKafkaProducer.KafkaTransactionContext}.
+ */
+ @VisibleForTesting
+ @Internal
+ public static class ContextStateSerializer extends TypeSerializerSingleton<FlinkKafkaProducer.KafkaTransactionContext> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionContext createInstance() {
+ return null;
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionContext copy(FlinkKafkaProducer.KafkaTransactionContext from) {
+ return from;
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionContext copy(
+ FlinkKafkaProducer.KafkaTransactionContext from,
+ FlinkKafkaProducer.KafkaTransactionContext reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(
+ FlinkKafkaProducer.KafkaTransactionContext record,
+ DataOutputView target) throws IOException {
+ int numIds = record.transactionalIds.size();
+ target.writeInt(numIds);
+ for (String id : record.transactionalIds) {
+ target.writeUTF(id);
+ }
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionContext deserialize(DataInputView source) throws IOException {
+ int numIds = source.readInt();
+ Set<String> ids = new HashSet<>(numIds);
+ for (int i = 0; i < numIds; i++) {
+ ids.add(source.readUTF());
+ }
+ return new FlinkKafkaProducer.KafkaTransactionContext(ids);
+ }
+
+ @Override
+ public FlinkKafkaProducer.KafkaTransactionContext deserialize(
+ FlinkKafkaProducer.KafkaTransactionContext reuse,
+ DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(
+ DataInputView source,
+ DataOutputView target) throws IOException {
+ int numIds = source.readInt();
+ target.writeInt(numIds);
+ for (int i = 0; i < numIds; i++) {
+ target.writeUTF(source.readUTF());
+ }
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof FlinkKafkaProducer.ContextStateSerializer;
+ }
+ }
+
+ /**
+ * Keep information required to deduce next safe to use transactional id.
+ */
+ public static class NextTransactionalIdHint {
+ public int lastParallelism = 0;
+ public long nextFreeTransactionalId = 0;
+
+ public NextTransactionalIdHint() {
+ this(0, 0);
+ }
+
+ public NextTransactionalIdHint(int parallelism, long nextFreeTransactionalId) {
+ this.lastParallelism = parallelism;
+ this.nextFreeTransactionalId = nextFreeTransactionalId;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
new file mode 100644
index 00000000000..e7379e0def3
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
@@ -0,0 +1,59 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.types.Row;
+
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Kafka table sink for writing data into Kafka.
+ */
+@Internal
+public class KafkaTableSink extends KafkaTableSinkBase {
+
+ public KafkaTableSink(
+ TableSchema schema,
+ String topic,
+ Properties properties,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
+ SerializationSchema<Row> serializationSchema) {
+
+ super(schema, topic, properties, partitioner, serializationSchema);
+ }
+
+ @Override
+ protected SinkFunction<Row> createKafkaProducer(
+ String topic,
+ Properties properties,
+ SerializationSchema<Row> serializationSchema,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
+ return new FlinkKafkaProducer<>(
+ topic,
+ new KeyedSerializationSchemaWrapper<>(serializationSchema),
+ properties,
+ partitioner);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
new file mode 100644
index 00000000000..f309536776e
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
@@ -0,0 +1,103 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Kafka {@link StreamTableSource}.
+ */
+@Internal
+public class KafkaTableSource extends KafkaTableSourceBase {
+
+ /**
+ * Creates a generic Kafka {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute.
+ * @param rowtimeAttributeDescriptors Descriptor for a rowtime attribute
+ * @param fieldMapping Mapping for the fields of the table schema to
+ * fields of the physical returned type.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ * @param startupMode Startup mode for the contained consumer.
+ * @param specificStartupOffsets Specific startup offsets; only relevant when startup
+ * mode is {@link StartupMode#SPECIFIC_OFFSETS}.
+ */
+ public KafkaTableSource(
+ TableSchema schema,
+ Optional<String> proctimeAttribute,
+ List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
+ Optional<Map<String, String>> fieldMapping,
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema,
+ StartupMode startupMode,
+ Map<KafkaTopicPartition, Long> specificStartupOffsets) {
+
+ super(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ properties,
+ deserializationSchema,
+ startupMode,
+ specificStartupOffsets);
+ }
+
+ /**
+ * Creates a generic Kafka {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ */
+ public KafkaTableSource(
+ TableSchema schema,
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema) {
+
+ super(schema, topic, properties, deserializationSchema);
+ }
+
+ @Override
+ protected FlinkKafkaConsumerBase<Row> createKafkaConsumer(
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema) {
+
+ return new FlinkKafkaConsumer<Row>(topic, deserializationSchema, properties);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactory.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactory.java
new file mode 100644
index 00000000000..b0dfc54e671
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactory.java
@@ -0,0 +1,89 @@
+/*
+ * 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.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.descriptors.KafkaValidator;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Factory for creating configured instances of {@link KafkaTableSource}.
+ */
+public class KafkaTableSourceSinkFactory extends KafkaTableSourceSinkFactoryBase {
+
+ @Override
+ protected String kafkaVersion() {
+ return KafkaValidator.CONNECTOR_VERSION_VALUE_20;
+ }
+
+ @Override
+ protected boolean supportsKafkaTimestamps() {
+ return true;
+ }
+
+ @Override
+ protected KafkaTableSourceBase createKafkaTableSource(
+ TableSchema schema,
+ Optional<String> proctimeAttribute,
+ List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
+ Map<String, String> fieldMapping,
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema,
+ StartupMode startupMode,
+ Map<KafkaTopicPartition, Long> specificStartupOffsets) {
+
+ return new KafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ Optional.of(fieldMapping),
+ topic,
+ properties,
+ deserializationSchema,
+ startupMode,
+ specificStartupOffsets);
+ }
+
+ @Override
+ protected KafkaTableSinkBase createKafkaTableSink(
+ TableSchema schema,
+ String topic,
+ Properties properties,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
+ SerializationSchema<Row> serializationSchema) {
+
+ return new KafkaTableSink(
+ schema,
+ topic,
+ properties,
+ partitioner,
+ serializationSchema);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaInternalProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaInternalProducer.java
new file mode 100644
index 00000000000..ea6d9c4ba6d
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaInternalProducer.java
@@ -0,0 +1,270 @@
+/*
+ * 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.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.clients.producer.internals.TransactionalRequestResult;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Internal flink kafka producer.
+ */
+@PublicEvolving
+public class FlinkKafkaInternalProducer<K, V> implements Producer<K, V> {
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaInternalProducer.class);
+
+ protected final KafkaProducer<K, V> kafkaProducer;
+
+ @Nullable
+ protected final String transactionalId;
+
+ public FlinkKafkaInternalProducer(Properties properties) {
+ transactionalId = properties.getProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
+ kafkaProducer = new KafkaProducer<>(properties);
+ }
+
+ // -------------------------------- Simple proxy method calls --------------------------------
+
+ @Override
+ public void initTransactions() {
+ kafkaProducer.initTransactions();
+ }
+
+ @Override
+ public void beginTransaction() throws ProducerFencedException {
+ kafkaProducer.beginTransaction();
+ }
+
+ @Override
+ public void commitTransaction() throws ProducerFencedException {
+ kafkaProducer.commitTransaction();
+ }
+
+ @Override
+ public void abortTransaction() throws ProducerFencedException {
+ kafkaProducer.abortTransaction();
+ }
+
+ @Override
+ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException {
+ kafkaProducer.sendOffsetsToTransaction(offsets, consumerGroupId);
+ }
+
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
+ return kafkaProducer.send(record);
+ }
+
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+ return kafkaProducer.send(record, callback);
+ }
+
+ @Override
+ public List<PartitionInfo> partitionsFor(String topic) {
+ return kafkaProducer.partitionsFor(topic);
+ }
+
+ @Override
+ public Map<MetricName, ? extends Metric> metrics() {
+ return kafkaProducer.metrics();
+ }
+
+ @Override
+ public void close() {
+ kafkaProducer.close();
+ }
+
+ @Override
+ public void close(long timeout, TimeUnit unit) {
+ kafkaProducer.close(timeout, unit);
+ }
+
+ // -------------------------------- New methods or methods with changed behaviour --------------------------------
+
+ @Override
+ public void flush() {
+ kafkaProducer.flush();
+ if (transactionalId != null) {
+ flushNewPartitions();
+ }
+ }
+
+ /**
+ * Instead of obtaining producerId and epoch from the transaction coordinator, re-use previously obtained ones,
+ * so that we can resume transaction after a restart. Implementation of this method is based on
+ * {@link KafkaProducer#initTransactions}.
+ * https://github.com/apache/kafka/commit/5d2422258cb975a137a42a4e08f03573c49a387e#diff-f4ef1afd8792cd2a2e9069cd7ddea630
+ */
+ public void resumeTransaction(long producerId, short epoch) {
+ Preconditions.checkState(producerId >= 0 && epoch >= 0, "Incorrect values for producerId {} and epoch {}", producerId, epoch);
+ LOG.info("Attempting to resume transaction {} with producerId {} and epoch {}", transactionalId, producerId, epoch);
+
+ Object transactionManager = getValue(kafkaProducer, "transactionManager");
+ synchronized (transactionManager) {
+ Object nextSequence = getValue(transactionManager, "nextSequence");
+
+ invoke(transactionManager, "transitionTo", getEnum("org.apache.kafka.clients.producer.internals.TransactionManager$State.INITIALIZING"));
+ invoke(nextSequence, "clear");
+
+ Object producerIdAndEpoch = getValue(transactionManager, "producerIdAndEpoch");
+ setValue(producerIdAndEpoch, "producerId", producerId);
+ setValue(producerIdAndEpoch, "epoch", epoch);
+
+ invoke(transactionManager, "transitionTo", getEnum("org.apache.kafka.clients.producer.internals.TransactionManager$State.READY"));
+
+ invoke(transactionManager, "transitionTo", getEnum("org.apache.kafka.clients.producer.internals.TransactionManager$State.IN_TRANSACTION"));
+ setValue(transactionManager, "transactionStarted", true);
+ }
+ }
+
+ public String getTransactionalId() {
+ return transactionalId;
+ }
+
+ public long getProducerId() {
+ Object transactionManager = getValue(kafkaProducer, "transactionManager");
+ Object producerIdAndEpoch = getValue(transactionManager, "producerIdAndEpoch");
+ return (long) getValue(producerIdAndEpoch, "producerId");
+ }
+
+ public short getEpoch() {
+ Object transactionManager = getValue(kafkaProducer, "transactionManager");
+ Object producerIdAndEpoch = getValue(transactionManager, "producerIdAndEpoch");
+ return (short) getValue(producerIdAndEpoch, "epoch");
+ }
+
+ @VisibleForTesting
+ public int getTransactionCoordinatorId() {
+ Object transactionManager = getValue(kafkaProducer, "transactionManager");
+ Node node = (Node) invoke(transactionManager, "coordinator", FindCoordinatorRequest.CoordinatorType.TRANSACTION);
+ return node.id();
+ }
+
+ /**
+ * Besides committing {@link org.apache.kafka.clients.producer.KafkaProducer#commitTransaction} is also adding new
+ * partitions to the transaction. flushNewPartitions method is moving this logic to pre-commit/flush, to make
+ * resumeTransaction simpler. Otherwise resumeTransaction would require to restore state of the not yet added/"in-flight"
+ * partitions.
+ */
+ private void flushNewPartitions() {
+ LOG.info("Flushing new partitions");
+ TransactionalRequestResult result = enqueueNewPartitions();
+ Object sender = getValue(kafkaProducer, "sender");
+ invoke(sender, "wakeup");
+ result.await();
+ }
+
+ private TransactionalRequestResult enqueueNewPartitions() {
+ Object transactionManager = getValue(kafkaProducer, "transactionManager");
+ synchronized (transactionManager) {
+ Object txnRequestHandler = invoke(transactionManager, "addPartitionsToTransactionHandler");
+ invoke(transactionManager, "enqueueRequest", new Class[]{txnRequestHandler.getClass().getSuperclass()}, new Object[]{txnRequestHandler});
+ TransactionalRequestResult result = (TransactionalRequestResult) getValue(txnRequestHandler, txnRequestHandler.getClass().getSuperclass(), "result");
+ return result;
+ }
+ }
+
+ protected static Enum<?> getEnum(String enumFullName) {
+ String[] x = enumFullName.split("\\.(?=[^\\.]+$)");
+ if (x.length == 2) {
+ String enumClassName = x[0];
+ String enumName = x[1];
+ try {
+ Class<Enum> cl = (Class<Enum>) Class.forName(enumClassName);
+ return Enum.valueOf(cl, enumName);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Incompatible KafkaProducer version", e);
+ }
+ }
+ return null;
+ }
+
+ protected static Object invoke(Object object, String methodName, Object... args) {
+ Class<?>[] argTypes = new Class[args.length];
+ for (int i = 0; i < args.length; i++) {
+ argTypes[i] = args[i].getClass();
+ }
+ return invoke(object, methodName, argTypes, args);
+ }
+
+ private static Object invoke(Object object, String methodName, Class<?>[] argTypes, Object[] args) {
+ try {
+ Method method = object.getClass().getDeclaredMethod(methodName, argTypes);
+ method.setAccessible(true);
+ return method.invoke(object, args);
+ } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
+ throw new RuntimeException("Incompatible KafkaProducer version", e);
+ }
+ }
+
+ protected static Object getValue(Object object, String fieldName) {
+ return getValue(object, object.getClass(), fieldName);
+ }
+
+ private static Object getValue(Object object, Class<?> clazz, String fieldName) {
+ try {
+ Field field = clazz.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return field.get(object);
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new RuntimeException("Incompatible KafkaProducer version", e);
+ }
+ }
+
+ protected static void setValue(Object object, String fieldName, Object value) {
+ try {
+ Field field = object.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(object, value);
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new RuntimeException("Incompatible KafkaProducer version", e);
+ }
+ }
+
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Handover.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Handover.java
new file mode 100644
index 00000000000..02cbd9223f5
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Handover.java
@@ -0,0 +1,218 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.io.Closeable;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The Handover is a utility to hand over data (a buffer of records) and exception from a
+ * <i>producer</i> thread to a <i>consumer</i> thread. It effectively behaves like a
+ * "size one blocking queue", with some extras around exception reporting, closing, and
+ * waking up thread without {@link Thread#interrupt() interrupting} threads.
+ *
+ * <p>This class is used in the Flink Kafka Consumer to hand over data and exceptions between
+ * the thread that runs the KafkaConsumer class and the main thread.
+ *
+ * <p>The Handover has the notion of "waking up" the producer thread with a {@link WakeupException}
+ * rather than a thread interrupt.
+ *
+ * <p>The Handover can also be "closed", signalling from one thread to the other that it
+ * the thread has terminated.
+ */
+@ThreadSafe
+@Internal
+public final class Handover implements Closeable {
+
+ private final Object lock = new Object();
+
+ private ConsumerRecords<byte[], byte[]> next;
+ private Throwable error;
+ private boolean wakeupProducer;
+
+ /**
+ * Polls the next element from the Handover, possibly blocking until the next element is
+ * available. This method behaves similar to polling from a blocking queue.
+ *
+ * <p>If an exception was handed in by the producer ({@link #reportError(Throwable)}), then
+ * that exception is thrown rather than an element being returned.
+ *
+ * @return The next element (buffer of records, never null).
+ *
+ * @throws ClosedException Thrown if the Handover was {@link #close() closed}.
+ * @throws Exception Rethrows exceptions from the {@link #reportError(Throwable)} method.
+ */
+ @Nonnull
+ public ConsumerRecords<byte[], byte[]> pollNext() throws Exception {
+ synchronized (lock) {
+ while (next == null && error == null) {
+ lock.wait();
+ }
+
+ ConsumerRecords<byte[], byte[]> n = next;
+ if (n != null) {
+ next = null;
+ lock.notifyAll();
+ return n;
+ }
+ else {
+ ExceptionUtils.rethrowException(error, error.getMessage());
+
+ // this statement cannot be reached since the above method always throws an exception
+ // this is only here to silence the compiler and any warnings
+ return ConsumerRecords.empty();
+ }
+ }
+ }
+
+ /**
+ * Hands over an element from the producer. If the Handover already has an element that was
+ * not yet picked up by the consumer thread, this call blocks until the consumer picks up that
+ * previous element.
+ *
+ * <p>This behavior is similar to a "size one" blocking queue.
+ *
+ * @param element The next element to hand over.
+ *
+ * @throws InterruptedException
+ * Thrown, if the thread is interrupted while blocking for the Handover to be empty.
+ * @throws WakeupException
+ * Thrown, if the {@link #wakeupProducer()} method is called while blocking for
+ * the Handover to be empty.
+ * @throws ClosedException
+ * Thrown if the Handover was closed or concurrently being closed.
+ */
+ public void produce(final ConsumerRecords<byte[], byte[]> element)
+ throws InterruptedException, WakeupException, ClosedException {
+
+ checkNotNull(element);
+
+ synchronized (lock) {
+ while (next != null && !wakeupProducer) {
+ lock.wait();
+ }
+
+ wakeupProducer = false;
+
+ // if there is still an element, we must have been woken up
+ if (next != null) {
+ throw new WakeupException();
+ }
+ // if there is no error, then this is open and can accept this element
+ else if (error == null) {
+ next = element;
+ lock.notifyAll();
+ }
+ // an error marks this as closed for the producer
+ else {
+ throw new ClosedException();
+ }
+ }
+ }
+
+ /**
+ * Reports an exception. The consumer will throw the given exception immediately, if
+ * it is currently blocked in the {@link #pollNext()} method, or the next time it
+ * calls that method.
+ *
+ * <p>After this method has been called, no call to either {@link #produce(ConsumerRecords)}
+ * or {@link #pollNext()} will ever return regularly any more, but will always return
+ * exceptionally.
+ *
+ * <p>If another exception was already reported, this method does nothing.
+ *
+ * <p>For the producer, the Handover will appear as if it was {@link #close() closed}.
+ *
+ * @param t The exception to report.
+ */
+ public void reportError(Throwable t) {
+ checkNotNull(t);
+
+ synchronized (lock) {
+ // do not override the initial exception
+ if (error == null) {
+ error = t;
+ }
+ next = null;
+ lock.notifyAll();
+ }
+ }
+
+ /**
+ * Closes the handover. Both the {@link #produce(ConsumerRecords)} method and the
+ * {@link #pollNext()} will throw a {@link ClosedException} on any currently blocking and
+ * future invocations.
+ *
+ * <p>If an exception was previously reported via the {@link #reportError(Throwable)} method,
+ * that exception will not be overridden. The consumer thread will throw that exception upon
+ * calling {@link #pollNext()}, rather than the {@code ClosedException}.
+ */
+ @Override
+ public void close() {
+ synchronized (lock) {
+ next = null;
+ wakeupProducer = false;
+
+ if (error == null) {
+ error = new ClosedException();
+ }
+ lock.notifyAll();
+ }
+ }
+
+ /**
+ * Wakes the producer thread up. If the producer thread is currently blocked in
+ * the {@link #produce(ConsumerRecords)} method, it will exit the method throwing
+ * a {@link WakeupException}.
+ */
+ public void wakeupProducer() {
+ synchronized (lock) {
+ wakeupProducer = true;
+ lock.notifyAll();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * An exception thrown by the Handover in the {@link #pollNext()} or
+ * {@link #produce(ConsumerRecords)} method, after the Handover was closed via
+ * {@link #close()}.
+ */
+ public static final class ClosedException extends Exception {
+ private static final long serialVersionUID = 1L;
+ }
+
+ /**
+ * A special exception thrown bv the Handover in the {@link #produce(ConsumerRecords)}
+ * method when the producer is woken up from a blocking call via {@link #wakeupProducer()}.
+ */
+ public static final class WakeupException extends Exception {
+ private static final long serialVersionUID = 1L;
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java
new file mode 100644
index 00000000000..2b44bb96064
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java
@@ -0,0 +1,520 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.streaming.connectors.kafka.internals.ClosableBlockingQueue;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaCommitCallback;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionState;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionStateSentinel;
+import org.apache.flink.streaming.connectors.kafka.internals.metrics.KafkaMetricWrapper;
+
+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.clients.consumer.OffsetCommitCallback;
+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 javax.annotation.Nonnull;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The thread the runs the {@link KafkaConsumer}, connecting to the brokers and polling records.
+ * The thread pushes the data into a {@link Handover} to be picked up by the fetcher that will
+ * deserialize and emit the records.
+ *
+ * <p><b>IMPORTANT:</b> This thread must not be interrupted when attempting to shut it down.
+ * The Kafka consumer code was found to not always handle interrupts well, and to even
+ * deadlock in certain situations.
+ *
+ * <p>Implementation Note: This code is written to be reusable in later versions of the KafkaConsumer.
+ * Because Kafka is not maintaining binary compatibility, we use a "call bridge" as an indirection
+ * to the KafkaConsumer calls that change signature.
+ */
+@Internal
+public class KafkaConsumerThread extends Thread {
+
+ /** Logger for this consumer. */
+ private final Logger log;
+
+ /** The handover of data and exceptions between the consumer thread and the task thread. */
+ private final Handover handover;
+
+ /** The next offsets that the main thread should commit and the commit callback. */
+ private final AtomicReference<Tuple2<Map<TopicPartition, OffsetAndMetadata>, KafkaCommitCallback>> nextOffsetsToCommit;
+
+ /** The configuration for the Kafka consumer. */
+ private final Properties kafkaProperties;
+
+ /** The queue of unassigned partitions that we need to assign to the Kafka consumer. */
+ private final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue;
+
+ /** The maximum number of milliseconds to wait for a fetch batch. */
+ private final long pollTimeout;
+
+ /** Flag whether to add Kafka's metrics to the Flink metrics. */
+ private final boolean useMetrics;
+
+ /**
+ * @deprecated We should only be publishing to the {{@link #consumerMetricGroup}}.
+ * This is kept to retain compatibility for metrics.
+ **/
+ @Deprecated
+ private final MetricGroup subtaskMetricGroup;
+
+ /** We get this from the outside to publish metrics. */
+ private final MetricGroup consumerMetricGroup;
+
+ /** Reference to the Kafka consumer, once it is created. */
+ private volatile KafkaConsumer<byte[], byte[]> consumer;
+
+ /** This lock is used to isolate the consumer for partition reassignment. */
+ private final Object consumerReassignmentLock;
+
+ /** Indication if this consumer has any assigned partition. */
+ private boolean hasAssignedPartitions;
+
+ /**
+ * Flag to indicate whether an external operation ({@link #setOffsetsToCommit(Map, KafkaCommitCallback)}
+ * or {@link #shutdown()}) had attempted to wakeup the consumer while it was isolated for partition reassignment.
+ */
+ private volatile boolean hasBufferedWakeup;
+
+ /** Flag to mark the main work loop as alive. */
+ private volatile boolean running;
+
+ /** Flag tracking whether the latest commit request has completed. */
+ private volatile boolean commitInProgress;
+
+ public KafkaConsumerThread(
+ Logger log,
+ Handover handover,
+ Properties kafkaProperties,
+ ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue,
+ String threadName,
+ long pollTimeout,
+ boolean useMetrics,
+ MetricGroup consumerMetricGroup,
+ MetricGroup subtaskMetricGroup) {
+
+ super(threadName);
+ setDaemon(true);
+
+ this.log = checkNotNull(log);
+ this.handover = checkNotNull(handover);
+ this.kafkaProperties = checkNotNull(kafkaProperties);
+ this.consumerMetricGroup = checkNotNull(consumerMetricGroup);
+ this.subtaskMetricGroup = checkNotNull(subtaskMetricGroup);
+
+ this.unassignedPartitionsQueue = checkNotNull(unassignedPartitionsQueue);
+
+ this.pollTimeout = pollTimeout;
+ this.useMetrics = useMetrics;
+
+ this.consumerReassignmentLock = new Object();
+ this.nextOffsetsToCommit = new AtomicReference<>();
+ this.running = true;
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void run() {
+ // early exit check
+ if (!running) {
+ return;
+ }
+
+ // this is the means to talk to FlinkKafkaConsumer's main thread
+ final Handover handover = this.handover;
+
+ // 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.
+ try {
+ this.consumer = getConsumer(kafkaProperties);
+ }
+ catch (Throwable t) {
+ handover.reportError(t);
+ return;
+ }
+
+ // from here on, the consumer is guaranteed to be closed properly
+ try {
+ // register Kafka's very own metrics in Flink's metric reporters
+ if (useMetrics) {
+ // register Kafka metrics to Flink
+ 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 Kafka metrics, register them
+ for (Map.Entry<MetricName, ? extends Metric> metric: metrics.entrySet()) {
+ consumerMetricGroup.gauge(metric.getKey().name(), new KafkaMetricWrapper(metric.getValue()));
+
+ // TODO this metric is kept for compatibility purposes; should remove in the future
+ subtaskMetricGroup.gauge(metric.getKey().name(), new KafkaMetricWrapper(metric.getValue()));
+ }
+ }
+ }
+
+ // early exit check
+ if (!running) {
+ return;
+ }
+
+ // the latest bulk of records. May carry across the loop if the thread is woken up
+ // from blocking on the handover
+ ConsumerRecords<byte[], byte[]> records = null;
+
+ // reused variable to hold found unassigned new partitions.
+ // found partitions are not carried across loops using this variable;
+ // they are carried across via re-adding them to the unassigned partitions queue
+ List<KafkaTopicPartitionState<TopicPartition>> newPartitions;
+
+ // main fetch loop
+ while (running) {
+
+ // check if there is something to commit
+ if (!commitInProgress) {
+ // get and reset the work-to-be committed, so we don't repeatedly commit the same
+ final Tuple2<Map<TopicPartition, OffsetAndMetadata>, KafkaCommitCallback> commitOffsetsAndCallback =
+ nextOffsetsToCommit.getAndSet(null);
+
+ if (commitOffsetsAndCallback != null) {
+ log.debug("Sending async offset commit request to Kafka broker");
+
+ // also record that a commit is already in progress
+ // the order here matters! first set the flag, then send the commit command.
+ commitInProgress = true;
+ consumer.commitAsync(commitOffsetsAndCallback.f0, new CommitCallback(commitOffsetsAndCallback.f1));
+ }
+ }
+
+ try {
+ if (hasAssignedPartitions) {
+ newPartitions = unassignedPartitionsQueue.pollBatch();
+ }
+ else {
+ // if no assigned partitions block until we get at least one
+ // instead of hot spinning this loop. We rely on a fact that
+ // unassignedPartitionsQueue will be closed on a shutdown, so
+ // we don't block indefinitely
+ newPartitions = unassignedPartitionsQueue.getBatchBlocking();
+ }
+ if (newPartitions != null) {
+ reassignPartitions(newPartitions);
+ }
+ } catch (AbortedReassignmentException e) {
+ continue;
+ }
+
+ if (!hasAssignedPartitions) {
+ // Without assigned partitions KafkaConsumer.poll will throw an exception
+ continue;
+ }
+
+ // get the next batch of records, unless we did not manage to hand the old batch over
+ if (records == null) {
+ try {
+ records = consumer.poll(pollTimeout);
+ }
+ catch (WakeupException we) {
+ continue;
+ }
+ }
+
+ try {
+ handover.produce(records);
+ records = null;
+ }
+ catch (Handover.WakeupException e) {
+ // fall through the loop
+ }
+ }
+ // end main fetch loop
+ }
+ catch (Throwable t) {
+ // let the main thread know and exit
+ // it may be that this exception comes because the main thread closed the handover, in
+ // which case the below reporting is irrelevant, but does not hurt either
+ handover.reportError(t);
+ }
+ finally {
+ // make sure the handover is closed if it is not already closed or has an error
+ handover.close();
+
+ // make sure the KafkaConsumer is closed
+ try {
+ consumer.close();
+ }
+ catch (Throwable t) {
+ log.warn("Error while closing Kafka consumer", t);
+ }
+ }
+ }
+
+ /**
+ * Shuts this thread down, waking up the thread gracefully if blocked (without Thread.interrupt() calls).
+ */
+ public void shutdown() {
+ running = false;
+
+ // wake up all blocking calls on the queue
+ unassignedPartitionsQueue.close();
+
+ // We cannot call close() on the KafkaConsumer, because it will actually throw
+ // an exception if a concurrent call is in progress
+
+ // this wakes up the consumer if it is blocked handing over records
+ handover.wakeupProducer();
+
+ // this wakes up the consumer if it is blocked in a kafka poll
+ synchronized (consumerReassignmentLock) {
+ if (consumer != null) {
+ consumer.wakeup();
+ } else {
+ // the consumer is currently isolated for partition reassignment;
+ // set this flag so that the wakeup state is restored once the reassignment is complete
+ hasBufferedWakeup = true;
+ }
+ }
+ }
+
+ /**
+ * Tells this thread to commit a set of offsets. This method does not block, the committing
+ * operation will happen asynchronously.
+ *
+ * <p>Only one commit operation may be pending at any time. If the committing takes longer than
+ * the frequency with which this method is called, then some commits may be skipped due to being
+ * superseded by newer ones.
+ *
+ * @param offsetsToCommit The offsets to commit
+ * @param commitCallback callback when Kafka commit completes
+ */
+ void setOffsetsToCommit(
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ @Nonnull KafkaCommitCallback commitCallback) {
+
+ // record the work to be committed by the main consumer thread and make sure the consumer notices that
+ if (nextOffsetsToCommit.getAndSet(Tuple2.of(offsetsToCommit, commitCallback)) != null) {
+ log.warn("Committing offsets to Kafka takes longer than the checkpoint interval. " +
+ "Skipping commit of previous offsets because newer complete checkpoint offsets are available. " +
+ "This does not compromise Flink's checkpoint integrity.");
+ }
+
+ // if the consumer is blocked in a poll() or handover operation, wake it up to commit soon
+ handover.wakeupProducer();
+
+ synchronized (consumerReassignmentLock) {
+ if (consumer != null) {
+ consumer.wakeup();
+ } else {
+ // the consumer is currently isolated for partition reassignment;
+ // set this flag so that the wakeup state is restored once the reassignment is complete
+ hasBufferedWakeup = true;
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Reestablishes the assigned partitions for the consumer.
+ * The reassigned partitions consists of the provided new partitions and whatever partitions
+ * was already previously assigned to the consumer.
+ *
+ * <p>The reassignment process is protected against wakeup calls, so that after
+ * this method returns, the consumer is either untouched or completely reassigned
+ * with the correct offset positions.
+ *
+ * <p>If the consumer was already woken-up prior to a reassignment resulting in an
+ * interruption any time during the reassignment, the consumer is guaranteed
+ * to roll back as if it was untouched. On the other hand, if there was an attempt
+ * to wakeup the consumer during the reassignment, the wakeup call is "buffered"
+ * until the reassignment completes.
+ *
+ * <p>This method is exposed for testing purposes.
+ */
+ @VisibleForTesting
+ void reassignPartitions(List<KafkaTopicPartitionState<TopicPartition>> newPartitions) throws Exception {
+ if (newPartitions.size() == 0) {
+ return;
+ }
+ hasAssignedPartitions = true;
+ boolean reassignmentStarted = false;
+
+ // since the reassignment may introduce several Kafka blocking calls that cannot be interrupted,
+ // the consumer needs to be isolated from external wakeup calls in setOffsetsToCommit() and shutdown()
+ // until the reassignment is complete.
+ final KafkaConsumer<byte[], byte[]> consumerTmp;
+ synchronized (consumerReassignmentLock) {
+ consumerTmp = this.consumer;
+ this.consumer = null;
+ }
+
+ final Map<TopicPartition, Long> oldPartitionAssignmentsToPosition = new HashMap<>();
+ try {
+ for (TopicPartition oldPartition : consumerTmp.assignment()) {
+ oldPartitionAssignmentsToPosition.put(oldPartition, consumerTmp.position(oldPartition));
+ }
+
+ final List<TopicPartition> newPartitionAssignments =
+ new ArrayList<>(newPartitions.size() + oldPartitionAssignmentsToPosition.size());
+ newPartitionAssignments.addAll(oldPartitionAssignmentsToPosition.keySet());
+ newPartitionAssignments.addAll(convertKafkaPartitions(newPartitions));
+
+ // reassign with the new partitions
+ consumerTmp.assign(newPartitionAssignments);
+ reassignmentStarted = true;
+
+ // old partitions should be seeked to their previous position
+ for (Map.Entry<TopicPartition, Long> oldPartitionToPosition : oldPartitionAssignmentsToPosition.entrySet()) {
+ consumerTmp.seek(oldPartitionToPosition.getKey(), oldPartitionToPosition.getValue());
+ }
+
+ // offsets in the state of new partitions may still be placeholder sentinel values if we are:
+ // (1) starting fresh,
+ // (2) checkpoint / savepoint state we were restored with had not completely
+ // been replaced with actual offset values yet, or
+ // (3) the partition was newly discovered after startup;
+ // replace those with actual offsets, according to what the sentinel value represent.
+ for (KafkaTopicPartitionState<TopicPartition> newPartitionState : newPartitions) {
+ if (newPartitionState.getOffset() == KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET) {
+ consumerTmp.seekToBeginning(Collections.singletonList(newPartitionState.getKafkaPartitionHandle()));
+ newPartitionState.setOffset(consumerTmp.position(newPartitionState.getKafkaPartitionHandle()) - 1);
+ } else if (newPartitionState.getOffset() == KafkaTopicPartitionStateSentinel.LATEST_OFFSET) {
+ consumerTmp.seekToEnd(Collections.singletonList(newPartitionState.getKafkaPartitionHandle()));
+ newPartitionState.setOffset(consumerTmp.position(newPartitionState.getKafkaPartitionHandle()) - 1);
+ } else if (newPartitionState.getOffset() == KafkaTopicPartitionStateSentinel.GROUP_OFFSET) {
+ // the KafkaConsumer by default will automatically seek the consumer position
+ // to the committed group offset, so we do not need to do it.
+
+ newPartitionState.setOffset(consumerTmp.position(newPartitionState.getKafkaPartitionHandle()) - 1);
+ } else {
+ consumerTmp.seek(newPartitionState.getKafkaPartitionHandle(), newPartitionState.getOffset() + 1);
+ }
+ }
+ } catch (WakeupException e) {
+ // a WakeupException may be thrown if the consumer was invoked wakeup()
+ // before it was isolated for the reassignment. In this case, we abort the
+ // reassignment and just re-expose the original consumer.
+
+ synchronized (consumerReassignmentLock) {
+ this.consumer = consumerTmp;
+
+ // if reassignment had already started and affected the consumer,
+ // we do a full roll back so that it is as if it was left untouched
+ if (reassignmentStarted) {
+ this.consumer.assign(new ArrayList<>(oldPartitionAssignmentsToPosition.keySet()));
+
+ for (Map.Entry<TopicPartition, Long> oldPartitionToPosition : oldPartitionAssignmentsToPosition.entrySet()) {
+ this.consumer.seek(oldPartitionToPosition.getKey(), oldPartitionToPosition.getValue());
+ }
+ }
+
+ // no need to restore the wakeup state in this case,
+ // since only the last wakeup call is effective anyways
+ hasBufferedWakeup = false;
+
+ // re-add all new partitions back to the unassigned partitions queue to be picked up again
+ for (KafkaTopicPartitionState<TopicPartition> newPartition : newPartitions) {
+ unassignedPartitionsQueue.add(newPartition);
+ }
+
+ // this signals the main fetch loop to continue through the loop
+ throw new AbortedReassignmentException();
+ }
+ }
+
+ // reassignment complete; expose the reassigned consumer
+ synchronized (consumerReassignmentLock) {
+ this.consumer = consumerTmp;
+
+ // restore wakeup state for the consumer if necessary
+ if (hasBufferedWakeup) {
+ this.consumer.wakeup();
+ hasBufferedWakeup = false;
+ }
+ }
+ }
+
+ @VisibleForTesting
+ KafkaConsumer<byte[], byte[]> getConsumer(Properties kafkaProperties) {
+ return new KafkaConsumer<>(kafkaProperties);
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ private static List<TopicPartition> convertKafkaPartitions(List<KafkaTopicPartitionState<TopicPartition>> partitions) {
+ ArrayList<TopicPartition> result = new ArrayList<>(partitions.size());
+ for (KafkaTopicPartitionState<TopicPartition> p : partitions) {
+ result.add(p.getKafkaPartitionHandle());
+ }
+ return result;
+ }
+
+ private class CommitCallback implements OffsetCommitCallback {
+
+ private final KafkaCommitCallback internalCommitCallback;
+
+ CommitCallback(KafkaCommitCallback internalCommitCallback) {
+ this.internalCommitCallback = checkNotNull(internalCommitCallback);
+ }
+
+ @Override
+ public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception ex) {
+ commitInProgress = false;
+
+ if (ex != null) {
+ log.warn("Committing offsets to Kafka failed. This does not compromise Flink's checkpoints.", ex);
+ internalCommitCallback.onException(ex);
+ } else {
+ internalCommitCallback.onSuccess();
+ }
+ }
+ }
+
+ /**
+ * Utility exception that serves as a signal for the main loop to continue through the loop
+ * if a reassignment attempt was aborted due to an pre-reassignment wakeup call on the consumer.
+ */
+ private static class AbortedReassignmentException extends Exception {
+ private static final long serialVersionUID = 1L;
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaFetcher.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaFetcher.java
new file mode 100644
index 00000000000..59e63dd44f7
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaFetcher.java
@@ -0,0 +1,233 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.metrics.MetricGroup;
+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;
+import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaCommitCallback;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionState;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.util.SerializedValue;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A fetcher that fetches data from Kafka brokers via the Kafka consumer API.
+ *
+ * @param <T> The type of elements produced by the fetcher.
+ */
+@Internal
+public class KafkaFetcher<T> extends AbstractFetcher<T, TopicPartition> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaFetcher.class);
+
+ // ------------------------------------------------------------------------
+
+ /** The schema to convert between Kafka's byte messages, and Flink's objects. */
+ private final KeyedDeserializationSchema<T> deserializer;
+
+ /** The handover of data and exceptions between the consumer thread and the task thread. */
+ private final Handover handover;
+
+ /** The thread that runs the actual KafkaConsumer and hand the record batches to this fetcher. */
+ private final KafkaConsumerThread consumerThread;
+
+ /** Flag to mark the main work loop as alive. */
+ private volatile boolean running = true;
+
+ // ------------------------------------------------------------------------
+
+ public KafkaFetcher(
+ SourceFunction.SourceContext<T> sourceContext,
+ Map<KafkaTopicPartition, Long> assignedPartitionsWithInitialOffsets,
+ SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
+ SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
+ ProcessingTimeService processingTimeProvider,
+ long autoWatermarkInterval,
+ ClassLoader userCodeClassLoader,
+ String taskNameWithSubtasks,
+ KeyedDeserializationSchema<T> deserializer,
+ Properties kafkaProperties,
+ long pollTimeout,
+ MetricGroup subtaskMetricGroup,
+ MetricGroup consumerMetricGroup,
+ boolean useMetrics) throws Exception {
+ super(
+ sourceContext,
+ assignedPartitionsWithInitialOffsets,
+ watermarksPeriodic,
+ watermarksPunctuated,
+ processingTimeProvider,
+ autoWatermarkInterval,
+ userCodeClassLoader,
+ consumerMetricGroup,
+ useMetrics);
+
+ this.deserializer = deserializer;
+ this.handover = new Handover();
+
+ this.consumerThread = new KafkaConsumerThread(
+ LOG,
+ handover,
+ kafkaProperties,
+ unassignedPartitionsQueue,
+ getFetcherName() + " for " + taskNameWithSubtasks,
+ pollTimeout,
+ useMetrics,
+ consumerMetricGroup,
+ subtaskMetricGroup);
+ }
+
+ // ------------------------------------------------------------------------
+ // Fetcher work methods
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void runFetchLoop() throws Exception {
+ try {
+ final Handover handover = this.handover;
+
+ // kick off the actual Kafka consumer
+ consumerThread.start();
+
+ while (running) {
+ // this blocks until we get the next records
+ // it automatically re-throws exceptions encountered in the consumer thread
+ final ConsumerRecords<byte[], byte[]> records = handover.pollNext();
+
+ // get the records for each topic partition
+ for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitionStates()) {
+
+ List<ConsumerRecord<byte[], byte[]>> partitionRecords =
+ records.records(partition.getKafkaPartitionHandle());
+
+ for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
+ final 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 updates offset state atomically
+ // and deals with timestamps and watermark generation
+ emitRecord(value, partition, record.offset(), record);
+ }
+ }
+ }
+ }
+ finally {
+ // this signals the consumer thread that no more work is to be done
+ consumerThread.shutdown();
+ }
+
+ // on a clean exit, wait for the runner thread
+ try {
+ consumerThread.join();
+ }
+ catch (InterruptedException e) {
+ // may be the result of a wake-up interruption after an exception.
+ // we ignore this here and only restore the interruption state
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ // flag the main thread to exit. A thread interrupt will come anyways.
+ running = false;
+ handover.close();
+ consumerThread.shutdown();
+ }
+
+ protected void emitRecord(
+ T record,
+ KafkaTopicPartitionState<TopicPartition> partition,
+ long offset,
+ ConsumerRecord<?, ?> consumerRecord) throws Exception {
+
+ emitRecordWithTimestamp(record, partition, offset, consumerRecord.timestamp());
+ }
+
+ /**
+ * Gets the name of this fetcher, for thread naming and logging purposes.
+ */
+ protected String getFetcherName() {
+ return "Kafka Fetcher";
+ }
+
+ // ------------------------------------------------------------------------
+ // Implement Methods of the AbstractFetcher
+ // ------------------------------------------------------------------------
+
+ @Override
+ public TopicPartition createKafkaPartitionHandle(KafkaTopicPartition partition) {
+ return new TopicPartition(partition.getTopic(), partition.getPartition());
+ }
+
+ @Override
+ protected void doCommitInternalOffsetsToKafka(
+ Map<KafkaTopicPartition, Long> offsets,
+ @Nonnull KafkaCommitCallback commitCallback) throws Exception {
+
+ @SuppressWarnings("unchecked")
+ List<KafkaTopicPartitionState<TopicPartition>> partitions = subscribedPartitionStates();
+
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>(partitions.size());
+
+ for (KafkaTopicPartitionState<TopicPartition> partition : partitions) {
+ Long lastProcessedOffset = offsets.get(partition.getKafkaTopicPartition());
+ if (lastProcessedOffset != null) {
+ checkState(lastProcessedOffset >= 0, "Illegal offset value to commit");
+
+ // committed offsets through the KafkaConsumer need to be 1 more than the last processed offset.
+ // This does not affect Flink's checkpoints/saved state.
+ long offsetToCommit = lastProcessedOffset + 1;
+
+ offsetsToCommit.put(partition.getKafkaPartitionHandle(), new OffsetAndMetadata(offsetToCommit));
+ partition.setCommittedOffset(offsetToCommit);
+ }
+ }
+
+ // record the work to be committed by the main consumer thread and make sure the consumer notices that
+ consumerThread.setOffsetsToCommit(offsetsToCommit, commitCallback);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaPartitionDiscoverer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaPartitionDiscoverer.java
new file mode 100644
index 00000000000..cc705ea94be
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaPartitionDiscoverer.java
@@ -0,0 +1,106 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicsDescriptor;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.PartitionInfo;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A partition discoverer that can be used to discover topics and partitions metadata
+ * from Kafka brokers via the Kafka high-level consumer API.
+ */
+@Internal
+public class KafkaPartitionDiscoverer extends AbstractPartitionDiscoverer {
+
+ private final Properties kafkaProperties;
+
+ private KafkaConsumer<?, ?> kafkaConsumer;
+
+ public KafkaPartitionDiscoverer(
+ KafkaTopicsDescriptor topicsDescriptor,
+ int indexOfThisSubtask,
+ int numParallelSubtasks,
+ Properties kafkaProperties) {
+
+ super(topicsDescriptor, indexOfThisSubtask, numParallelSubtasks);
+ this.kafkaProperties = checkNotNull(kafkaProperties);
+ }
+
+ @Override
+ protected void initializeConnections() {
+ this.kafkaConsumer = new KafkaConsumer<>(kafkaProperties);
+ }
+
+ @Override
+ protected List<String> getAllTopics() throws AbstractPartitionDiscoverer.WakeupException {
+ try {
+ return new ArrayList<>(kafkaConsumer.listTopics().keySet());
+ } catch (org.apache.kafka.common.errors.WakeupException e) {
+ // rethrow our own wakeup exception
+ throw new AbstractPartitionDiscoverer.WakeupException();
+ }
+ }
+
+ @Override
+ protected List<KafkaTopicPartition> getAllPartitionsForTopics(List<String> topics) throws AbstractPartitionDiscoverer.WakeupException {
+ List<KafkaTopicPartition> partitions = new LinkedList<>();
+
+ try {
+ for (String topic : topics) {
+ for (PartitionInfo partitionInfo : kafkaConsumer.partitionsFor(topic)) {
+ partitions.add(new KafkaTopicPartition(partitionInfo.topic(), partitionInfo.partition()));
+ }
+ }
+ } catch (org.apache.kafka.common.errors.WakeupException e) {
+ // rethrow our own wakeup exception
+ throw new AbstractPartitionDiscoverer.WakeupException();
+ }
+
+ return partitions;
+ }
+
+ @Override
+ protected void wakeupConnections() {
+ if (this.kafkaConsumer != null) {
+ this.kafkaConsumer.wakeup();
+ }
+ }
+
+ @Override
+ protected void closeConnections() throws Exception {
+ if (this.kafkaConsumer != null) {
+ this.kafkaConsumer.close();
+
+ // de-reference the consumer to avoid closing multiple times
+ this.kafkaConsumer = null;
+ }
+ }
+
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java
new file mode 100644
index 00000000000..ffebf568c0d
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java
@@ -0,0 +1,96 @@
+/*
+ * 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.annotation.Internal;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Class responsible for generating transactional ids to use when communicating with Kafka.
+ *
+ * <p>It guarantees that:
+ * <ul>
+ * <li>generated ids to use will never clash with ids to use from different subtasks
+ * <li>generated ids to abort will never clash with ids to abort from different subtasks
+ * <li>generated ids to use will never clash with ids to abort from different subtasks
+ * </ul>
+ * In other words, any particular generated id will always be assigned to one and only one subtask.
+ */
+@Internal
+public class TransactionalIdsGenerator {
+ private final String prefix;
+ private final int subtaskIndex;
+ private final int totalNumberOfSubtasks;
+ private final int poolSize;
+ private final int safeScaleDownFactor;
+
+ public TransactionalIdsGenerator(
+ String prefix,
+ int subtaskIndex,
+ int totalNumberOfSubtasks,
+ int poolSize,
+ int safeScaleDownFactor) {
+ checkArgument(subtaskIndex < totalNumberOfSubtasks);
+ checkArgument(poolSize > 0);
+ checkArgument(safeScaleDownFactor > 0);
+ checkArgument(subtaskIndex >= 0);
+
+ this.prefix = checkNotNull(prefix);
+ this.subtaskIndex = subtaskIndex;
+ this.totalNumberOfSubtasks = totalNumberOfSubtasks;
+ this.poolSize = poolSize;
+ this.safeScaleDownFactor = safeScaleDownFactor;
+ }
+
+ /**
+ * Range of available transactional ids to use is:
+ * [nextFreeTransactionalId, nextFreeTransactionalId + parallelism * kafkaProducersPoolSize)
+ * loop below picks in a deterministic way a subrange of those available transactional ids based on index of
+ * this subtask.
+ */
+ public Set<String> generateIdsToUse(long nextFreeTransactionalId) {
+ Set<String> transactionalIds = new HashSet<>();
+ for (int i = 0; i < poolSize; i++) {
+ long transactionalId = nextFreeTransactionalId + subtaskIndex * poolSize + i;
+ transactionalIds.add(generateTransactionalId(transactionalId));
+ }
+ return transactionalIds;
+ }
+
+ /**
+ * If we have to abort previous transactional id in case of restart after a failure BEFORE first checkpoint
+ * completed, we don't know what was the parallelism used in previous attempt. In that case we must guess the ids
+ * range to abort based on current configured pool size, current parallelism and safeScaleDownFactor.
+ */
+ public Set<String> generateIdsToAbort() {
+ Set<String> idsToAbort = new HashSet<>();
+ for (int i = 0; i < safeScaleDownFactor; i++) {
+ idsToAbort.addAll(generateIdsToUse(i * poolSize * totalNumberOfSubtasks));
+ }
+ return idsToAbort;
+ }
+
+ private String generateTransactionalId(long transactionalId) {
+ return String.format(prefix + "-%d", transactionalId);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMutableWrapper.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMutableWrapper.java
new file mode 100644
index 00000000000..39f4df2d050
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMutableWrapper.java
@@ -0,0 +1,45 @@
+/*
+ * 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.metrics;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Gauge;
+
+import org.apache.kafka.common.Metric;
+
+/**
+ * Gauge for getting the current value of a Kafka metric.
+ */
+@Internal
+public class KafkaMetricMutableWrapper implements Gauge<Double> {
+ private Metric kafkaMetric;
+
+ public KafkaMetricMutableWrapper(Metric metric) {
+ this.kafkaMetric = metric;
+ }
+
+ @Override
+ public Double getValue() {
+ return kafkaMetric.value();
+ }
+
+ public void setKafkaMetric(Metric kafkaMetric) {
+ this.kafkaMetric = kafkaMetric;
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory b/flink-connectors/flink-connector-kafka/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
new file mode 100644
index 00000000000..f945958741c
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactory
diff --git a/flink-connectors/flink-connector-kafka/src/main/resources/log4j.properties b/flink-connectors/flink-connector-kafka/src/main/resources/log4j.properties
new file mode 100644
index 00000000000..6eef1747ddf
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/resources/log4j.properties
@@ -0,0 +1,28 @@
+################################################################################
+# 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
+
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaInternalProducerITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaInternalProducerITCase.java
new file mode 100644
index 00000000000..d35af108ca4
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaInternalProducerITCase.java
@@ -0,0 +1,114 @@
+/*
+ * 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.internal.FlinkKafkaInternalProducer;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
+
+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.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Properties;
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for our own {@link FlinkKafkaInternalProducer}.
+ */
+@SuppressWarnings("serial")
+public class FlinkKafkaInternalProducerITCase extends KafkaTestBase {
+ protected String transactionalId;
+ protected Properties extraProperties;
+
+ @Before
+ public void before() {
+ transactionalId = UUID.randomUUID().toString();
+ extraProperties = new Properties();
+ extraProperties.putAll(standardProps);
+ extraProperties.put("transactional.id", transactionalId);
+ extraProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ extraProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ extraProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+ extraProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+ extraProperties.put("isolation.level", "read_committed");
+ }
+
+ @Test(timeout = 30000L)
+ public void testHappyPath() throws IOException {
+ String topicName = "flink-kafka-producer-happy-path";
+ try (Producer<String, String> kafkaProducer = new FlinkKafkaInternalProducer<>(extraProperties)) {
+ kafkaProducer.initTransactions();
+ kafkaProducer.beginTransaction();
+ kafkaProducer.send(new ProducerRecord<>(topicName, "42", "42"));
+ kafkaProducer.commitTransaction();
+ }
+ assertRecord(topicName, "42", "42");
+ deleteTestTopic(topicName);
+ }
+
+ @Test(timeout = 30000L)
+ public void testResumeTransaction() throws IOException {
+ String topicName = "flink-kafka-producer-resume-transaction";
+ try (FlinkKafkaInternalProducer<String, String> kafkaProducer = new FlinkKafkaInternalProducer<>(extraProperties)) {
+ kafkaProducer.initTransactions();
+ kafkaProducer.beginTransaction();
+ kafkaProducer.send(new ProducerRecord<>(topicName, "42", "42"));
+ kafkaProducer.flush();
+ long producerId = kafkaProducer.getProducerId();
+ short epoch = kafkaProducer.getEpoch();
+
+ try (FlinkKafkaInternalProducer<String, String> resumeProducer = new FlinkKafkaInternalProducer<>(extraProperties)) {
+ resumeProducer.resumeTransaction(producerId, epoch);
+ resumeProducer.commitTransaction();
+ }
+
+ assertRecord(topicName, "42", "42");
+
+ // this shouldn't throw - in case of network split, old producer might attempt to commit it's transaction
+ kafkaProducer.commitTransaction();
+
+ // this shouldn't fail also, for same reason as above
+ try (FlinkKafkaInternalProducer<String, String> resumeProducer = new FlinkKafkaInternalProducer<>(extraProperties)) {
+ resumeProducer.resumeTransaction(producerId, epoch);
+ resumeProducer.commitTransaction();
+ }
+ }
+ deleteTestTopic(topicName);
+ }
+
+ private void assertRecord(String topicName, String expectedKey, String expectedValue) {
+ try (KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(extraProperties)) {
+ kafkaConsumer.subscribe(Collections.singletonList(topicName));
+ ConsumerRecords<String, String> records = kafkaConsumer.poll(10000);
+
+ ConsumerRecord<String, String> record = Iterables.getOnlyElement(records);
+ assertEquals(expectedKey, record.key());
+ assertEquals(expectedValue, record.value());
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerITCase.java
new file mode 100644
index 00000000000..b14b3e2524b
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerITCase.java
@@ -0,0 +1,647 @@
+/*
+ * 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.ExecutionConfig;
+import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.StateObjectCollection;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
+
+import kafka.server.KafkaServer;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.util.ExceptionUtils.findThrowable;
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.hamcrest.Matchers.lessThan;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+
+/**
+ * IT cases for the {@link FlinkKafkaProducer}.
+ */
+public class FlinkKafkaProducerITCase extends KafkaTestBase {
+
+ protected String transactionalId;
+ protected Properties extraProperties;
+
+ protected TypeInformationSerializationSchema<Integer> integerSerializationSchema =
+ new TypeInformationSerializationSchema<>(BasicTypeInfo.INT_TYPE_INFO, new ExecutionConfig());
+ protected KeyedSerializationSchema<Integer> integerKeyedSerializationSchema =
+ new KeyedSerializationSchemaWrapper<>(integerSerializationSchema);
+
+ @Before
+ public void before() {
+ transactionalId = UUID.randomUUID().toString();
+ extraProperties = new Properties();
+ extraProperties.putAll(standardProps);
+ extraProperties.put("transactional.id", transactionalId);
+ extraProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ extraProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ extraProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+ extraProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+ extraProperties.put("isolation.level", "read_committed");
+ }
+
+ @Test
+ public void resourceCleanUpNone() throws Exception {
+ resourceCleanUp(FlinkKafkaProducer.Semantic.NONE);
+ }
+
+ @Test
+ public void resourceCleanUpAtLeastOnce() throws Exception {
+ resourceCleanUp(FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);
+ }
+
+ /**
+ * This tests checks whether there is some resource leak in form of growing threads number.
+ */
+ public void resourceCleanUp(FlinkKafkaProducer.Semantic semantic) throws Exception {
+ String topic = "flink-kafka-producer-resource-cleanup-" + semantic;
+
+ final int allowedEpsilonThreadCountGrow = 50;
+
+ Optional<Integer> initialActiveThreads = Optional.empty();
+ for (int i = 0; i < allowedEpsilonThreadCountGrow * 2; i++) {
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness1 =
+ createTestHarness(topic, 1, 1, 0, semantic)) {
+ testHarness1.setup();
+ testHarness1.open();
+ }
+
+ if (initialActiveThreads.isPresent()) {
+ assertThat("active threads count",
+ Thread.activeCount(),
+ lessThan(initialActiveThreads.get() + allowedEpsilonThreadCountGrow));
+ }
+ else {
+ initialActiveThreads = Optional.of(Thread.activeCount());
+ }
+ }
+ }
+
+ /**
+ * This test ensures that transactions reusing transactional.ids (after returning to the pool) will not clash
+ * with previous transactions using same transactional.ids.
+ */
+ @Test
+ public void testRestoreToCheckpointAfterExceedingProducersPool() throws Exception {
+ String topic = "flink-kafka-producer-fail-before-notify";
+
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness1 = createTestHarness(topic)) {
+ testHarness1.setup();
+ testHarness1.open();
+ testHarness1.processElement(42, 0);
+ OperatorSubtaskState snapshot = testHarness1.snapshot(0, 0);
+ testHarness1.processElement(43, 0);
+ testHarness1.notifyOfCompletedCheckpoint(0);
+ try {
+ for (int i = 0; i < FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE; i++) {
+ testHarness1.snapshot(i + 1, 0);
+ testHarness1.processElement(i, 0);
+ }
+ throw new IllegalStateException("This should not be reached.");
+ }
+ catch (Exception ex) {
+ if (!isCausedBy(FlinkKafkaErrorCode.PRODUCERS_POOL_EMPTY, ex)) {
+ throw ex;
+ }
+ }
+
+ // Resume transactions before testHarness1 is being closed (in case of failures close() might not be called)
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness2 = createTestHarness(topic)) {
+ testHarness2.setup();
+ // restore from snapshot1, transactions with records 43 and 44 should be aborted
+ testHarness2.initializeState(snapshot);
+ testHarness2.open();
+ }
+
+ assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42), 30_000L);
+ deleteTestTopic(topic);
+ }
+ catch (Exception ex) {
+ // testHarness1 will be fenced off after creating and closing testHarness2
+ if (!findThrowable(ex, ProducerFencedException.class).isPresent()) {
+ throw ex;
+ }
+ }
+ }
+
+ @Test
+ public void testFlinkKafkaProducer10FailBeforeNotify() throws Exception {
+ String topic = "flink-kafka-producer-fail-before-notify";
+
+ OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic);
+
+ testHarness.setup();
+ testHarness.open();
+ testHarness.processElement(42, 0);
+ testHarness.snapshot(0, 1);
+ testHarness.processElement(43, 2);
+ OperatorSubtaskState snapshot = testHarness.snapshot(1, 3);
+
+ int leaderId = kafkaServer.getLeaderToShutDown(topic);
+ failBroker(leaderId);
+
+ try {
+ testHarness.processElement(44, 4);
+ testHarness.snapshot(2, 5);
+ assertFalse(true);
+ }
+ catch (Exception ex) {
+ // expected
+ }
+ try {
+ testHarness.close();
+ }
+ catch (Exception ex) {
+ }
+
+ kafkaServer.restartBroker(leaderId);
+
+ testHarness = createTestHarness(topic);
+ testHarness.setup();
+ testHarness.initializeState(snapshot);
+ testHarness.close();
+
+ assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42, 43), 30_000L);
+
+ deleteTestTopic(topic);
+ }
+
+ @Test
+ public void testFlinkKafkaProducer10FailTransactionCoordinatorBeforeNotify() throws Exception {
+ String topic = "flink-kafka-producer-fail-transaction-coordinator-before-notify";
+
+ Properties properties = createProperties();
+
+ FlinkKafkaProducer<Integer> kafkaProducer = new FlinkKafkaProducer<>(
+ topic,
+ integerKeyedSerializationSchema,
+ properties,
+ FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
+
+ OneInputStreamOperatorTestHarness<Integer, Object> testHarness1 = new OneInputStreamOperatorTestHarness<>(
+ new StreamSink<>(kafkaProducer),
+ IntSerializer.INSTANCE);
+
+ testHarness1.setup();
+ testHarness1.open();
+ testHarness1.processElement(42, 0);
+ testHarness1.snapshot(0, 1);
+ testHarness1.processElement(43, 2);
+ int transactionCoordinatorId = kafkaProducer.getTransactionCoordinatorId();
+ OperatorSubtaskState snapshot = testHarness1.snapshot(1, 3);
+
+ failBroker(transactionCoordinatorId);
+
+ try {
+ testHarness1.processElement(44, 4);
+ testHarness1.notifyOfCompletedCheckpoint(1);
+ testHarness1.close();
+ }
+ catch (Exception ex) {
+ // Expected... some random exception could be thrown by any of the above operations.
+ }
+ finally {
+ kafkaServer.restartBroker(transactionCoordinatorId);
+ }
+
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness2 = createTestHarness(topic)) {
+ testHarness2.setup();
+ testHarness2.initializeState(snapshot);
+ testHarness2.open();
+ }
+
+ assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42, 43), 30_000L);
+
+ deleteTestTopic(topic);
+ }
+
+ /**
+ * This tests checks whether FlinkKafkaProducer correctly aborts lingering transactions after a failure.
+ * If such transactions were left alone lingering it consumers would be unable to read committed records
+ * that were created after this lingering transaction.
+ */
+ @Test
+ public void testFailBeforeNotifyAndResumeWorkAfterwards() throws Exception {
+ String topic = "flink-kafka-producer-fail-before-notify";
+
+ OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic);
+
+ testHarness.setup();
+ testHarness.open();
+ testHarness.processElement(42, 0);
+ testHarness.snapshot(0, 1);
+ testHarness.processElement(43, 2);
+ OperatorSubtaskState snapshot1 = testHarness.snapshot(1, 3);
+
+ testHarness.processElement(44, 4);
+ testHarness.snapshot(2, 5);
+ testHarness.processElement(45, 6);
+
+ // do not close previous testHarness to make sure that closing do not clean up something (in case of failure
+ // there might not be any close)
+ testHarness = createTestHarness(topic);
+ testHarness.setup();
+ // restore from snapshot1, transactions with records 44 and 45 should be aborted
+ testHarness.initializeState(snapshot1);
+ testHarness.open();
+
+ // write and commit more records, after potentially lingering transactions
+ testHarness.processElement(46, 7);
+ testHarness.snapshot(4, 8);
+ testHarness.processElement(47, 9);
+ testHarness.notifyOfCompletedCheckpoint(4);
+
+ //now we should have:
+ // - records 42 and 43 in committed transactions
+ // - aborted transactions with records 44 and 45
+ // - committed transaction with record 46
+ // - pending transaction with record 47
+ assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42, 43, 46), 30_000L);
+
+ testHarness.close();
+ deleteTestTopic(topic);
+ }
+
+ @Test
+ public void testFailAndRecoverSameCheckpointTwice() throws Exception {
+ String topic = "flink-kafka-producer-fail-and-recover-same-checkpoint-twice";
+
+ OperatorSubtaskState snapshot1;
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic)) {
+ testHarness.setup();
+ testHarness.open();
+ testHarness.processElement(42, 0);
+ testHarness.snapshot(0, 1);
+ testHarness.processElement(43, 2);
+ snapshot1 = testHarness.snapshot(1, 3);
+
+ testHarness.processElement(44, 4);
+ }
+
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic)) {
+ testHarness.setup();
+ // restore from snapshot1, transactions with records 44 and 45 should be aborted
+ testHarness.initializeState(snapshot1);
+ testHarness.open();
+
+ // write and commit more records, after potentially lingering transactions
+ testHarness.processElement(44, 7);
+ testHarness.snapshot(2, 8);
+ testHarness.processElement(45, 9);
+ }
+
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic)) {
+ testHarness.setup();
+ // restore from snapshot1, transactions with records 44 and 45 should be aborted
+ testHarness.initializeState(snapshot1);
+ testHarness.open();
+
+ // write and commit more records, after potentially lingering transactions
+ testHarness.processElement(44, 7);
+ testHarness.snapshot(3, 8);
+ testHarness.processElement(45, 9);
+ }
+
+ //now we should have:
+ // - records 42 and 43 in committed transactions
+ // - aborted transactions with records 44 and 45
+ assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42, 43), 30_000L);
+ deleteTestTopic(topic);
+ }
+
+ /**
+ * This tests checks whether FlinkKafkaProducer correctly aborts lingering transactions after a failure,
+ * which happened before first checkpoint and was followed up by reducing the parallelism.
+ * If such transactions were left alone lingering it consumers would be unable to read committed records
+ * that were created after this lingering transaction.
+ */
+ @Test
+ public void testScaleDownBeforeFirstCheckpoint() throws Exception {
+ String topic = "scale-down-before-first-checkpoint";
+
+ List<AutoCloseable> operatorsToClose = new ArrayList<>();
+ int preScaleDownParallelism = Math.max(2, FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR);
+ for (int subtaskIndex = 0; subtaskIndex < preScaleDownParallelism; subtaskIndex++) {
+ OneInputStreamOperatorTestHarness<Integer, Object> preScaleDownOperator = createTestHarness(
+ topic,
+ preScaleDownParallelism,
+ preScaleDownParallelism,
+ subtaskIndex,
+ FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
+
+ preScaleDownOperator.setup();
+ preScaleDownOperator.open();
+ preScaleDownOperator.processElement(subtaskIndex * 2, 0);
+ preScaleDownOperator.snapshot(0, 1);
+ preScaleDownOperator.processElement(subtaskIndex * 2 + 1, 2);
+
+ operatorsToClose.add(preScaleDownOperator);
+ }
+
+ // do not close previous testHarnesses to make sure that closing do not clean up something (in case of failure
+ // there might not be any close)
+
+ // After previous failure simulate restarting application with smaller parallelism
+ OneInputStreamOperatorTestHarness<Integer, Object> postScaleDownOperator1 = createTestHarness(topic, 1, 1, 0, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
+
+ postScaleDownOperator1.setup();
+ postScaleDownOperator1.open();
+
+ // write and commit more records, after potentially lingering transactions
+ postScaleDownOperator1.processElement(46, 7);
+ postScaleDownOperator1.snapshot(4, 8);
+ postScaleDownOperator1.processElement(47, 9);
+ postScaleDownOperator1.notifyOfCompletedCheckpoint(4);
+
+ //now we should have:
+ // - records 42, 43, 44 and 45 in aborted transactions
+ // - committed transaction with record 46
+ // - pending transaction with record 47
+ assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(46), 30_000L);
+
+ postScaleDownOperator1.close();
+ // ignore ProducerFencedExceptions, because postScaleDownOperator1 could reuse transactional ids.
+ for (AutoCloseable operatorToClose : operatorsToClose) {
+ closeIgnoringProducerFenced(operatorToClose);
+ }
+ deleteTestTopic(topic);
+ }
+
+ /**
+ * Each instance of FlinkKafkaProducer uses it's own pool of transactional ids. After the restore from checkpoint
+ * transactional ids are redistributed across the subtasks. In case of scale down, the surplus transactional ids
+ * are dropped. In case of scale up, new one are generated (for the new subtasks). This test make sure that sequence
+ * of scaling down and up again works fine. Especially it checks whether the newly generated ids in scaling up
+ * do not overlap with ids that were used before scaling down. For example we start with 4 ids and parallelism 4:
+ * [1], [2], [3], [4] - one assigned per each subtask
+ * we scale down to parallelism 2:
+ * [1, 2], [3, 4] - first subtask got id 1 and 2, second got ids 3 and 4
+ * surplus ids are dropped from the pools and we scale up to parallelism 3:
+ * [1 or 2], [3 or 4], [???]
+ * new subtask have to generate new id(s), but he can not use ids that are potentially in use, so it has to generate
+ * new ones that are greater then 4.
+ */
+ @Test
+ public void testScaleUpAfterScalingDown() throws Exception {
+ String topic = "scale-down-before-first-checkpoint";
+
+ final int parallelism1 = 4;
+ final int parallelism2 = 2;
+ final int parallelism3 = 3;
+ final int maxParallelism = Math.max(parallelism1, Math.max(parallelism2, parallelism3));
+
+ List<OperatorStateHandle> operatorSubtaskState = repartitionAndExecute(
+ topic,
+ Collections.emptyList(),
+ parallelism1,
+ maxParallelism,
+ IntStream.range(0, parallelism1).boxed().iterator());
+
+ operatorSubtaskState = repartitionAndExecute(
+ topic,
+ operatorSubtaskState,
+ parallelism2,
+ maxParallelism,
+ IntStream.range(parallelism1, parallelism1 + parallelism2).boxed().iterator());
+
+ operatorSubtaskState = repartitionAndExecute(
+ topic,
+ operatorSubtaskState,
+ parallelism3,
+ maxParallelism,
+ IntStream.range(parallelism1 + parallelism2, parallelism1 + parallelism2 + parallelism3).boxed().iterator());
+
+ // After each previous repartitionAndExecute call, we are left with some lingering transactions, that would
+ // not allow us to read all committed messages from the topic. Thus we initialize operators from
+ // OperatorSubtaskState once more, but without any new data. This should terminate all ongoing transactions.
+
+ operatorSubtaskState = repartitionAndExecute(
+ topic,
+ operatorSubtaskState,
+ 1,
+ maxParallelism,
+ Collections.emptyIterator());
+
+ assertExactlyOnceForTopic(
+ createProperties(),
+ topic,
+ 0,
+ IntStream.range(0, parallelism1 + parallelism2 + parallelism3).boxed().collect(Collectors.toList()),
+ 30_000L);
+ deleteTestTopic(topic);
+ }
+
+ private List<OperatorStateHandle> repartitionAndExecute(
+ String topic,
+ List<OperatorStateHandle> inputStates,
+ int parallelism,
+ int maxParallelism,
+ Iterator<Integer> inputData) throws Exception {
+
+ List<OperatorStateHandle> outputStates = new ArrayList<>();
+ List<OneInputStreamOperatorTestHarness<Integer, Object>> testHarnesses = new ArrayList<>();
+
+ for (int subtaskIndex = 0; subtaskIndex < parallelism; subtaskIndex++) {
+ OneInputStreamOperatorTestHarness<Integer, Object> testHarness =
+ createTestHarness(topic, maxParallelism, parallelism, subtaskIndex, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
+ testHarnesses.add(testHarness);
+
+ testHarness.setup();
+
+ testHarness.initializeState(new OperatorSubtaskState(
+ new StateObjectCollection<>(inputStates),
+ StateObjectCollection.empty(),
+ StateObjectCollection.empty(),
+ StateObjectCollection.empty()));
+ testHarness.open();
+
+ if (inputData.hasNext()) {
+ int nextValue = inputData.next();
+ testHarness.processElement(nextValue, 0);
+ OperatorSubtaskState snapshot = testHarness.snapshot(0, 0);
+
+ outputStates.addAll(snapshot.getManagedOperatorState());
+ checkState(snapshot.getRawOperatorState().isEmpty(), "Unexpected raw operator state");
+ checkState(snapshot.getManagedKeyedState().isEmpty(), "Unexpected managed keyed state");
+ checkState(snapshot.getRawKeyedState().isEmpty(), "Unexpected raw keyed state");
+
+ for (int i = 1; i < FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE - 1; i++) {
+ testHarness.processElement(-nextValue, 0);
+ testHarness.snapshot(i, 0);
+ }
+ }
+ }
+
+ for (OneInputStreamOperatorTestHarness<Integer, Object> testHarness : testHarnesses) {
+ testHarness.close();
+ }
+
+ return outputStates;
+ }
+
+ @Test
+ public void testRecoverCommittedTransaction() throws Exception {
+ String topic = "flink-kafka-producer-recover-committed-transaction";
+
+ OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic);
+
+ testHarness.setup();
+ testHarness.open(); // producerA - start transaction (txn) 0
+ testHarness.processElement(42, 0); // producerA - write 42 in txn 0
+ OperatorSubtaskState checkpoint0 = testHarness.snapshot(0, 1); // producerA - pre commit txn 0, producerB - start txn 1
+ testHarness.processElement(43, 2); // producerB - write 43 in txn 1
+ testHarness.notifyOfCompletedCheckpoint(0); // producerA - commit txn 0 and return to the pool
+ testHarness.snapshot(1, 3); // producerB - pre txn 1, producerA - start txn 2
+ testHarness.processElement(44, 4); // producerA - write 44 in txn 2
+ testHarness.close(); // producerA - abort txn 2
+
+ testHarness = createTestHarness(topic);
+ testHarness.initializeState(checkpoint0); // recover state 0 - producerA recover and commit txn 0
+ testHarness.close();
+
+ assertExactlyOnceForTopic(createProperties(), topic, 0, Arrays.asList(42), 30_000L);
+
+ deleteTestTopic(topic);
+ }
+
+ @Test
+ public void testRunOutOfProducersInThePool() throws Exception {
+ String topic = "flink-kafka-run-out-of-producers";
+
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness = createTestHarness(topic)) {
+
+ testHarness.setup();
+ testHarness.open();
+
+ for (int i = 0; i < FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE * 2; i++) {
+ testHarness.processElement(i, i * 2);
+ testHarness.snapshot(i, i * 2 + 1);
+ }
+ }
+ catch (Exception ex) {
+ if (!ex.getCause().getMessage().startsWith("Too many ongoing")) {
+ throw ex;
+ }
+ }
+ deleteTestTopic(topic);
+ }
+
+ // shut down a Kafka broker
+ private void failBroker(int brokerId) {
+ KafkaServer toShutDown = null;
+ for (KafkaServer server : kafkaServer.getBrokers()) {
+
+ if (kafkaServer.getBrokerId(server) == brokerId) {
+ toShutDown = server;
+ break;
+ }
+ }
+
+ if (toShutDown == null) {
+ StringBuilder listOfBrokers = new StringBuilder();
+ for (KafkaServer server : kafkaServer.getBrokers()) {
+ listOfBrokers.append(kafkaServer.getBrokerId(server));
+ listOfBrokers.append(" ; ");
+ }
+
+ throw new IllegalArgumentException("Cannot find broker to shut down: " + brokerId
+ + " ; available brokers: " + listOfBrokers.toString());
+ } else {
+ toShutDown.shutdown();
+ toShutDown.awaitShutdown();
+ }
+ }
+
+ private void closeIgnoringProducerFenced(AutoCloseable autoCloseable) throws Exception {
+ try {
+ autoCloseable.close();
+ }
+ catch (Exception ex) {
+ if (!(ex.getCause() instanceof ProducerFencedException)) {
+ throw ex;
+ }
+ }
+ }
+
+ private OneInputStreamOperatorTestHarness<Integer, Object> createTestHarness(String topic) throws Exception {
+ return createTestHarness(topic, 1, 1, 0, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
+ }
+
+ private OneInputStreamOperatorTestHarness<Integer, Object> createTestHarness(
+ String topic,
+ int maxParallelism,
+ int parallelism,
+ int subtaskIndex,
+ FlinkKafkaProducer.Semantic semantic) throws Exception {
+ Properties properties = createProperties();
+
+ FlinkKafkaProducer<Integer> kafkaProducer = new FlinkKafkaProducer<>(
+ topic,
+ integerKeyedSerializationSchema,
+ properties,
+ semantic);
+
+ return new OneInputStreamOperatorTestHarness<>(
+ new StreamSink<>(kafkaProducer),
+ maxParallelism,
+ parallelism,
+ subtaskIndex,
+ IntSerializer.INSTANCE,
+ new OperatorID(42, 44));
+ }
+
+ private Properties createProperties() {
+ Properties properties = new Properties();
+ properties.putAll(standardProps);
+ properties.putAll(secureProps);
+ properties.put(FlinkKafkaProducer.KEY_DISABLE_METRICS, "true");
+ return properties;
+ }
+
+ private boolean isCausedBy(FlinkKafkaErrorCode expectedErrorCode, Throwable ex) {
+ Optional<FlinkKafkaException> cause = findThrowable(ex, FlinkKafkaException.class);
+ if (cause.isPresent()) {
+ return cause.get().getErrorCode().equals(expectedErrorCode);
+ }
+ return false;
+ }
+
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerStateSerializerTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerStateSerializerTest.java
new file mode 100644
index 00000000000..adf66922d9c
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerStateSerializerTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
+import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.TransactionHolder;
+
+import java.util.Collections;
+import java.util.Optional;
+
+/**
+ * A test for the {@link TypeSerializer TypeSerializers} used for the Kafka producer state.
+ */
+public class FlinkKafkaProducerStateSerializerTest
+ extends SerializerTestBase<
+ TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>> {
+
+ @Override
+ protected TypeSerializer<
+ TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>> createSerializer() {
+ return new TwoPhaseCommitSinkFunction.StateSerializer<>(
+ new FlinkKafkaProducer.TransactionStateSerializer(),
+ new FlinkKafkaProducer.ContextStateSerializer());
+ }
+
+ @Override
+ protected Class<TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>> getTypeClass() {
+ return (Class) TwoPhaseCommitSinkFunction.State.class;
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>[] getTestData() {
+ //noinspection unchecked
+ return new TwoPhaseCommitSinkFunction.State[] {
+ new TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>(
+ new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+ Collections.emptyList(),
+ Optional.empty()),
+ new TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>(
+ new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 2711),
+ Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 42)),
+ Optional.empty()),
+ new TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>(
+ new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+ Collections.emptyList(),
+ Optional.of(new FlinkKafkaProducer.KafkaTransactionContext(Collections.emptySet()))),
+ new TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>(
+ new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+ Collections.emptyList(),
+ Optional.of(new FlinkKafkaProducer.KafkaTransactionContext(Collections.singleton("hello")))),
+ new TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>(
+ new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+ Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 0)),
+ Optional.of(new FlinkKafkaProducer.KafkaTransactionContext(Collections.emptySet()))),
+ new TwoPhaseCommitSinkFunction.State<
+ FlinkKafkaProducer.KafkaTransactionState,
+ FlinkKafkaProducer.KafkaTransactionContext>(
+ new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+ Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer.KafkaTransactionState("fake", 1L, (short) 42, null), 0)),
+ Optional.of(new FlinkKafkaProducer.KafkaTransactionContext(Collections.singleton("hello"))))
+ };
+ }
+
+ @Override
+ public void testInstantiate() {
+ // this serializer does not support instantiation
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
new file mode 100644
index 00000000000..11aee4a65fd
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
@@ -0,0 +1,353 @@
+/*
+ * 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.ExecutionConfig;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+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.FlinkKafkaPartitioner;
+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.junit.BeforeClass;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.util.Optional;
+
+/**
+ * IT cases for Kafka.
+ */
+public class KafkaITCase extends KafkaConsumerTestBase {
+
+ @BeforeClass
+ public static void prepare() throws ClassNotFoundException {
+ KafkaProducerTestBase.prepare();
+ ((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);
+ }
+
+ // ------------------------------------------------------------------------
+ // Suite of Tests
+ // ------------------------------------------------------------------------
+
+ @Test(timeout = 120000)
+ public void testFailOnNoBroker() throws Exception {
+ runFailOnNoBrokerTest();
+ }
+
+ @Test(timeout = 60000)
+ public void testConcurrentProducerConsumerTopology() throws Exception {
+ runSimpleConcurrentProducerConsumerTopology();
+ }
+
+ @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();
+ }
+
+ // --- 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 {
+ runEndOfStreamTest();
+ }
+
+ // --- startup mode ---
+
+ @Test(timeout = 60000)
+ public void testStartFromEarliestOffsets() throws Exception {
+ runStartFromEarliestOffsets();
+ }
+
+ @Test(timeout = 60000)
+ public void testStartFromLatestOffsets() throws Exception {
+ runStartFromLatestOffsets();
+ }
+
+ @Test(timeout = 60000)
+ public void testStartFromGroupOffsets() throws Exception {
+ runStartFromGroupOffsets();
+ }
+
+ @Test(timeout = 60000)
+ public void testStartFromSpecificOffsets() throws Exception {
+ runStartFromSpecificOffsets();
+ }
+
+ @Test(timeout = 60000)
+ public void testStartFromTimestamp() throws Exception {
+ runStartFromTimestamp();
+ }
+
+ // --- offset committing ---
+
+ @Test(timeout = 60000)
+ public void testCommitOffsetsToKafka() throws Exception {
+ runCommitOffsetsToKafka();
+ }
+
+ @Test(timeout = 60000)
+ public void testAutoOffsetRetrievalAndCommitToKafka() throws Exception {
+ runAutoOffsetRetrievalAndCommitToKafka();
+ }
+
+ /**
+ * Kafka 20 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.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.getConfig().disableSysoutLogging();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+ DataStream<Long> streamWithTimestamps = env.addSource(new SourceFunction<Long>() {
+ private static final long serialVersionUID = -2255115836471289626L;
+ boolean running = true;
+
+ @Override
+ public void run(SourceContext<Long> ctx) throws Exception {
+ long i = 0;
+ while (running) {
+ ctx.collectWithTimestamp(i, i * 2);
+ if (i++ == 1110L) {
+ running = false;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+ });
+
+ final TypeInformationSerializationSchema<Long> longSer = new TypeInformationSerializationSchema<>(Types.LONG, env.getConfig());
+ FlinkKafkaProducer<Long> prod = new FlinkKafkaProducer<>(topic, new KeyedSerializationSchemaWrapper<>(longSer), standardProps, Optional.of(new FlinkKafkaPartitioner<Long>() {
+ private static final long serialVersionUID = -6730989584364230617L;
+
+ @Override
+ public int partition(Long next, byte[] key, byte[] value, String targetTopic, int[] partitions) {
+ return (int) (next % 3);
+ }
+ }));
+ prod.setWriteTimestampToKafka(true);
+
+ streamWithTimestamps.addSink(prod).setParallelism(3);
+
+ env.execute("Produce some");
+
+ // ---------- Consume stream from Kafka -------------------
+
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.getConfig().disableSysoutLogging();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+ FlinkKafkaConsumer<Long> kafkaSource = new FlinkKafkaConsumer<>(topic, new KafkaITCase.LimitedLongDeserializer(), standardProps);
+ kafkaSource.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<Long>() {
+ private static final long serialVersionUID = -4834111173247835189L;
+
+ @Nullable
+ @Override
+ public Watermark checkAndGetNextWatermark(Long lastElement, long extractedTimestamp) {
+ if (lastElement % 11 == 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> {
+
+ private static final long serialVersionUID = 1353168781235526806L;
+
+ public TimestampValidatingOperator() {
+ super(new SinkFunction<Long>() {
+ private static final long serialVersionUID = -6676565693361786524L;
+
+ @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() % 11 != 0 && mark.getTimestamp() != Long.MAX_VALUE) {
+ throw new RuntimeException("Invalid watermark: " + mark.getTimestamp());
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ if (elCount != 1110L) {
+ 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 static final long serialVersionUID = 6966177118923713521L;
+ private final TypeInformation<Long> ti;
+ private final TypeSerializer<Long> ser;
+ long cnt = 0;
+
+ public LimitedLongDeserializer() {
+ this.ti = Types.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 > 1110L;
+ }
+ }
+
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerAtLeastOnceITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerAtLeastOnceITCase.java
new file mode 100644
index 00000000000..c07b84febdd
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerAtLeastOnceITCase.java
@@ -0,0 +1,44 @@
+/*
+ * 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.BeforeClass;
+
+/**
+ * IT cases for the {@link FlinkKafkaProducer}.
+ */
+@SuppressWarnings("serial")
+public class KafkaProducerAtLeastOnceITCase extends KafkaProducerTestBase {
+
+ @BeforeClass
+ public static void prepare() throws ClassNotFoundException {
+ KafkaProducerTestBase.prepare();
+ ((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);
+ }
+
+ @Override
+ public void testExactlyOnceRegularSink() throws Exception {
+ // disable test for at least once semantic
+ }
+
+ @Override
+ public void testExactlyOnceCustomOperator() throws Exception {
+ // disable test for at least once semantic
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExactlyOnceITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExactlyOnceITCase.java
new file mode 100644
index 00000000000..45a0312b2a7
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExactlyOnceITCase.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.kafka;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * IT cases for the {@link FlinkKafkaProducer}.
+ */
+@SuppressWarnings("serial")
+public class KafkaProducerExactlyOnceITCase extends KafkaProducerTestBase {
+ @BeforeClass
+ public static void prepare() throws ClassNotFoundException {
+ KafkaProducerTestBase.prepare();
+ ((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
+ }
+
+ @Override
+ public void testOneToOneAtLeastOnceRegularSink() throws Exception {
+ // TODO: fix this test
+ // currently very often (~50% cases) KafkaProducer live locks itself on commitTransaction call.
+ // Somehow Kafka 10 doesn't play along with NetworkFailureProxy. This can either mean a bug in Kafka
+ // that it doesn't work well with some weird network failures, or the NetworkFailureProxy is a broken design
+ // and this test should be reimplemented in completely different way...
+ }
+
+ @Override
+ public void testOneToOneAtLeastOnceCustomOperator() throws Exception {
+ // TODO: fix this test
+ // currently very often (~50% cases) KafkaProducer live locks itself on commitTransaction call.
+ // Somehow Kafka 10 doesn't play along with NetworkFailureProxy. This can either mean a bug in Kafka
+ // that it doesn't work well with some weird network failures, or the NetworkFailureProxy is a broken design
+ // and this test should be reimplemented in completely different way...
+ }
+
+ @Test
+ public void testMultipleSinkOperators() throws Exception {
+ testExactlyOnce(false, 2);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTest.java
new file mode 100644
index 00000000000..5043880b21e
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.descriptors.KafkaValidator;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Test for {@link KafkaTableSource} and {@link KafkaTableSink} created
+ * by {@link KafkaTableSourceSinkFactory}.
+ */
+public class KafkaTableSourceSinkFactoryTest extends KafkaTableSourceSinkFactoryTestBase {
+
+ @Override
+ protected String getKafkaVersion() {
+ return KafkaValidator.CONNECTOR_VERSION_VALUE_20;
+ }
+
+ @Override
+ protected Class<FlinkKafkaConsumerBase<Row>> getExpectedFlinkKafkaConsumer() {
+ return (Class) FlinkKafkaConsumer.class;
+ }
+
+ @Override
+ protected Class<?> getExpectedFlinkKafkaProducer() {
+ return FlinkKafkaProducer.class;
+ }
+
+ @Override
+ protected KafkaTableSourceBase getExpectedKafkaTableSource(
+ TableSchema schema,
+ Optional<String> proctimeAttribute,
+ List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
+ Map<String, String> fieldMapping,
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema,
+ StartupMode startupMode,
+ Map<KafkaTopicPartition, Long> specificStartupOffsets) {
+
+ return new KafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ Optional.of(fieldMapping),
+ topic,
+ properties,
+ deserializationSchema,
+ startupMode,
+ specificStartupOffsets);
+ }
+
+ @Override
+ protected KafkaTableSinkBase getExpectedKafkaTableSink(
+ TableSchema schema,
+ String topic,
+ Properties properties,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
+ SerializationSchema<Row> serializationSchema) {
+
+ return new KafkaTableSink(
+ schema,
+ topic,
+ properties,
+ partitioner,
+ serializationSchema);
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
new file mode 100644
index 00000000000..12a0c04cfb8
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -0,0 +1,440 @@
+/*
+ * 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.networking.NetworkFailuresProxy;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
+import org.apache.flink.util.NetUtils;
+
+import kafka.common.KafkaException;
+import kafka.metrics.KafkaMetricsReporter;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServer;
+import org.apache.commons.collections.list.UnmodifiableList;
+import org.apache.commons.io.FileUtils;
+import org.apache.curator.test.TestingServer;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.utils.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.net.BindException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.UUID;
+
+import scala.collection.mutable.ArraySeq;
+
+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.
+ */
+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 FlinkKafkaProducer.Semantic producerSemantic = FlinkKafkaProducer.Semantic.EXACTLY_ONCE;
+ // 6 seconds is default. Seems to be too small for travis. 30 seconds
+ private int zkTimeout = 30000;
+ private Config config;
+
+ public void setProducerSemantic(FlinkKafkaProducer.Semantic producerSemantic) {
+ this.producerSemantic = producerSemantic;
+ }
+
+ @Override
+ public void prepare(Config config) {
+ //increase the timeout since in Travis ZK connection takes long time for secure connection.
+ if (config.isSecureMode()) {
+ //run only one kafka server to avoid multiple ZK connections from many instances - Travis timeout
+ config.setKafkaServersNumber(1);
+ zkTimeout = zkTimeout * 15;
+ }
+ this.config = config;
+
+ 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<>(config.getKafkaServersNumber());
+ for (int i = 0; i < config.getKafkaServersNumber(); i++) {
+ File tmpDir = new File(tmpKafkaParent, "server-" + i);
+ assertTrue("cannot create kafka temp dir", tmpDir.mkdir());
+ tmpKafkaDirs.add(tmpDir);
+ }
+
+ zookeeper = null;
+ brokers = null;
+
+ try {
+ zookeeper = new TestingServer(-1, tmpZkDir);
+ zookeeperConnectionString = zookeeper.getConnectString();
+ LOG.info("Starting Zookeeper with zookeeperConnectionString: {}", zookeeperConnectionString);
+
+ LOG.info("Starting KafkaServer");
+ brokers = new ArrayList<>(config.getKafkaServersNumber());
+
+ ListenerName listenerName = ListenerName.forSecurityProtocol(config.isSecureMode() ? SecurityProtocol.SASL_PLAINTEXT : SecurityProtocol.PLAINTEXT);
+ for (int i = 0; i < config.getKafkaServersNumber(); i++) {
+ KafkaServer kafkaServer = getKafkaServer(i, tmpKafkaDirs.get(i));
+ brokers.add(kafkaServer);
+ brokerConnectionString += hostAndPortToUrlString(KAFKA_HOST, kafkaServer.socketServer().boundPort(listenerName));
+ brokerConnectionString += ",";
+ }
+
+ 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("enable.auto.commit", "false");
+ 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.
+ standardProps.setProperty("max.partition.fetch.bytes", "256"); // make a lot of fetches (MESSAGES MUST BE SMALLER!)
+ }
+
+ @Override
+ public void deleteTestTopic(String topic) {
+ LOG.info("Deleting topic {}", topic);
+ try (AdminClient adminClient = AdminClient.create(getStandardProperties())) {
+ adminClient.deleteTopics(Collections.singleton(topic)).all().get();
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Delete test topic : " + topic + " failed, " + e.getMessage());
+ }
+ }
+
+ @Override
+ public void createTestTopic(String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
+ LOG.info("Creating topic {}", topic);
+ try (AdminClient adminClient = AdminClient.create(getStandardProperties())) {
+ NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
+ adminClient.createTopics(Collections.singleton(topicObj)).all().get();
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Create test topic : " + topic + " failed, " + e.getMessage());
+ }
+ }
+
+ @Override
+ public Properties getStandardProperties() {
+ return standardProps;
+ }
+
+ @Override
+ public Properties getSecureProperties() {
+ Properties prop = new Properties();
+ if (config.isSecureMode()) {
+ 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 getBrokerConnectionString() {
+ return brokerConnectionString;
+ }
+
+ @Override
+ public String getVersion() {
+ return "2.0";
+ }
+
+ @Override
+ public List<KafkaServer> getBrokers() {
+ return brokers;
+ }
+
+ @Override
+ public <T> FlinkKafkaConsumerBase<T> getConsumer(List<String> topics, KeyedDeserializationSchema<T> readSchema, Properties props) {
+ return new FlinkKafkaConsumer<T>(topics, readSchema, props);
+ }
+
+ @Override
+ public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(Properties properties, String topic, int partition, long timeout) {
+ List<ConsumerRecord<K, V>> result = new ArrayList<>();
+
+ try (KafkaConsumer<K, V> consumer = new KafkaConsumer<>(properties)) {
+ consumer.assign(Arrays.asList(new TopicPartition(topic, partition)));
+
+ while (true) {
+ boolean processedAtLeastOneRecord = false;
+
+ // wait for new records with timeout and break the loop if we didn't get any
+ Iterator<ConsumerRecord<K, V>> iterator = consumer.poll(timeout).iterator();
+ while (iterator.hasNext()) {
+ ConsumerRecord<K, V> record = iterator.next();
+ result.add(record);
+ processedAtLeastOneRecord = true;
+ }
+
+ if (!processedAtLeastOneRecord) {
+ break;
+ }
+ }
+ consumer.commitSync();
+ }
+
+ return UnmodifiableList.decorate(result);
+ }
+
+ @Override
+ public <T> StreamSink<T> getProducerSink(String topic, KeyedSerializationSchema<T> serSchema, Properties props, FlinkKafkaPartitioner<T> partitioner) {
+ return new StreamSink<>(new FlinkKafkaProducer<T>(
+ topic,
+ serSchema,
+ props,
+ Optional.ofNullable(partitioner),
+ producerSemantic,
+ FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE));
+ }
+
+ @Override
+ public <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props, FlinkKafkaPartitioner<T> partitioner) {
+ return stream.addSink(new FlinkKafkaProducer<T>(
+ topic,
+ serSchema,
+ props,
+ Optional.ofNullable(partitioner),
+ producerSemantic,
+ FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE));
+ }
+
+ @Override
+ public <T> DataStreamSink<T> writeToKafkaWithTimestamps(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props) {
+ FlinkKafkaProducer<T> prod = new FlinkKafkaProducer<T>(
+ topic,
+ serSchema,
+ props,
+ Optional.of(new FlinkFixedPartitioner<>()),
+ producerSemantic,
+ FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE);
+
+ prod.setWriteTimestampToKafka(true);
+
+ return stream.addSink(prod);
+ }
+
+ @Override
+ public KafkaOffsetHandler createOffsetHandler() {
+ return new KafkaOffsetHandlerImpl();
+ }
+
+ @Override
+ public void restartBroker(int leaderId) throws Exception {
+ brokers.set(leaderId, getKafkaServer(leaderId, tmpKafkaDirs.get(leaderId)));
+ }
+
+ @Override
+ public int getLeaderToShutDown(String topic) throws Exception {
+ AdminClient client = AdminClient.create(getStandardProperties());
+ TopicDescription result = client.describeTopics(Collections.singleton(topic)).all().get().get(topic);
+ return result.partitions().get(0).leader().id();
+ }
+
+ @Override
+ public int getBrokerId(KafkaServer server) {
+ return server.config().brokerId();
+ }
+
+ @Override
+ public boolean isSecureRunSupported() {
+ return true;
+ }
+
+ @Override
+ public void shutdown() throws Exception {
+ 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
+ }
+ }
+ }
+
+ 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));
+ kafkaProperties.put("transaction.max.timeout.ms", Integer.toString(1000 * 60 * 60 * 2)); // 2hours
+
+ // for CI stability, increase zookeeper session timeout
+ kafkaProperties.put("zookeeper.session.timeout.ms", zkTimeout);
+ kafkaProperties.put("zookeeper.connection.timeout.ms", zkTimeout);
+ if (config.getKafkaServerProperties() != null) {
+ kafkaProperties.putAll(config.getKafkaServerProperties());
+ }
+
+ final int numTries = 5;
+
+ for (int i = 1; i <= numTries; i++) {
+ int kafkaPort = NetUtils.getAvailablePort();
+ kafkaProperties.put("port", Integer.toString(kafkaPort));
+
+ if (config.isHideKafkaBehindProxy()) {
+ NetworkFailuresProxy proxy = createProxy(KAFKA_HOST, kafkaPort);
+ kafkaProperties.put("advertised.port", proxy.getLocalPort());
+ }
+
+ //to support secure kafka cluster
+ if (config.isSecureMode()) {
+ 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 {
+ scala.Option<String> stringNone = scala.Option.apply(null);
+ KafkaServer server = new KafkaServer(kafkaConfig, Time.SYSTEM, stringNone, new ArraySeq<KafkaMetricsReporter>(0));
+ 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.");
+ }
+
+ private class KafkaOffsetHandlerImpl implements KafkaOffsetHandler {
+
+ private final KafkaConsumer<byte[], byte[]> offsetClient;
+
+ public KafkaOffsetHandlerImpl() {
+ Properties props = new Properties();
+ props.putAll(standardProps);
+ props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+ props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+
+ offsetClient = new KafkaConsumer<>(props);
+ }
+
+ @Override
+ public Long getCommittedOffset(String topicName, int partition) {
+ OffsetAndMetadata committed = offsetClient.committed(new TopicPartition(topicName, partition));
+ return (committed != null) ? committed.offset() : null;
+ }
+
+ @Override
+ public void setCommittedOffset(String topicName, int partition, long offset) {
+ Map<TopicPartition, OffsetAndMetadata> partitionAndOffset = new HashMap<>();
+ partitionAndOffset.put(new TopicPartition(topicName, partition), new OffsetAndMetadata(offset));
+ offsetClient.commitSync(partitionAndOffset);
+ }
+
+ @Override
+ public void close() {
+ offsetClient.close();
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/resources/log4j-test.properties b/flink-connectors/flink-connector-kafka/src/test/resources/log4j-test.properties
new file mode 100644
index 00000000000..fbeb110350f
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/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
diff --git a/flink-connectors/pom.xml b/flink-connectors/pom.xml
index cacea91578e..d7a851a8063 100644
--- a/flink-connectors/pom.xml
+++ b/flink-connectors/pom.xml
@@ -56,6 +56,7 @@ under the License.
<module>flink-connector-nifi</module>
<module>flink-connector-cassandra</module>
<module>flink-connector-filesystem</module>
+ <module>flink-connector-kafka</module>
</modules>
<!-- override these root dependencies as 'provided', so they don't end up
diff --git a/pom.xml b/pom.xml
index 77094932682..0851fa033c0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -104,7 +104,7 @@ under the License.
<guava.version>18.0</guava.version>
<akka.version>2.4.20</akka.version>
<java.version>1.8</java.version>
- <slf4j.version>1.7.7</slf4j.version>
+ <slf4j.version>1.7.15</slf4j.version>
<log4j.version>1.2.17</log4j.version>
<!-- Overwrite default values from parent pom.
Intellij is (sometimes?) using those values to choose target language level
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services