[FLINK-8276] [kafka] Properly annotate APIs for Kafka connector

This closes apache#5173.
laubersder · Jan 12, 2018 · 4ceabed · 4ceabed
1 parent 9b5fce6
commit 4ceabed
Show file tree

Hide file tree

Showing 79 changed files with 177 additions and 0 deletions.
diff --git a/...0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java b/...0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java
@@ -59,6 +59,7 @@
  * <p>Please refer to Kafka's documentation for the available configuration properties:
  * http:https://kafka.apache.org/documentation.html#newconsumerconfigs</p>
  */
+@PublicEvolving
 public class FlinkKafkaConsumer010<T> extends FlinkKafkaConsumer09<T> {
 
  private static final long serialVersionUID = 2324564345203409112L;

diff --git a/...0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java b/...0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
@@ -39,6 +40,7 @@
 /**
  * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.10.x
  */
+@PublicEvolving
 public class FlinkKafkaProducer010<T> extends FlinkKafkaProducer09<T> {
 
  private static final long serialVersionUID = 1L;

diff --git a/...10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSource.java b/...10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSource.java
@@ -18,6 +18,7 @@
 
 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.table.api.TableSchema;
 import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
@@ -35,6 +36,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.10.
  */
+@PublicEvolving
 public class Kafka010AvroTableSource extends KafkaAvroTableSource {
 
  /**

diff --git a/...0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSink.java b/...0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSink.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
@@ -28,6 +29,7 @@
 /**
  * Kafka 0.10 {@link KafkaTableSink} that serializes data in JSON format.
  */
+@PublicEvolving
 public class Kafka010JsonTableSink extends KafkaJsonTableSink {
 
  /**

diff --git a/...10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java b/...10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java
@@ -18,6 +18,7 @@
 
 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.table.api.TableSchema;
 import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
@@ -32,6 +33,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.10.
  */
+@PublicEvolving
 public class Kafka010JsonTableSource extends KafkaJsonTableSource {
 
  /**

diff --git a/...a-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java b/...a-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java
@@ -18,6 +18,7 @@
 
 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.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.api.TableSchema;
@@ -29,6 +30,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.10.
  */
+@PublicEvolving
 public abstract class Kafka010TableSource extends KafkaTableSource {
 
  // The deserialization schema for the Kafka records

diff --git a/...0/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java b/...0/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka010Fetcher.java
@@ -18,6 +18,7 @@
 
 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;
@@ -42,6 +43,7 @@
  *
  * @param <T> The type of elements produced by the fetcher.
  */
+@Internal
 public class Kafka010Fetcher<T> extends Kafka09Fetcher<T> {
 
  public Kafka010Fetcher(

diff --git a/...ava/org/apache/flink/streaming/connectors/kafka/internal/Kafka010PartitionDiscoverer.java b/...ava/org/apache/flink/streaming/connectors/kafka/internal/Kafka010PartitionDiscoverer.java
@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.connectors.kafka.internal;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicsDescriptor;
 
 import java.util.Properties;
@@ -25,6 +26,7 @@
  * A partition discoverer that can be used to discover topics and partitions metadata
  * from Kafka brokers via the Kafka 0.10 high-level consumer API.
  */
+@Internal
 public class Kafka010PartitionDiscoverer extends Kafka09PartitionDiscoverer {
 
  public Kafka010PartitionDiscoverer(

diff --git a/...java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge010.java b/...java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerCallBridge010.java
@@ -18,6 +18,8 @@
 
 package org.apache.flink.streaming.connectors.kafka.internal;
 
+import org.apache.flink.annotation.Internal;
+
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
 
@@ -32,6 +34,7 @@
  *
  * <p>Because of that, we need two versions whose compiled code goes against different method signatures.
  */
+@Internal
 public class KafkaConsumerCallBridge010 extends KafkaConsumerCallBridge {
 
  @Override

diff --git a/....11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafka011ErrorCode.java b/....11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafka011ErrorCode.java
@@ -17,9 +17,12 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import org.apache.flink.annotation.PublicEvolving;
+
 /**
  * Error codes used in {@link FlinkKafka011Exception}.
  */
+@PublicEvolving
 public enum FlinkKafka011ErrorCode {
  PRODUCERS_POOL_EMPTY,
  EXTERNAL_ERROR

diff --git a/....11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafka011Exception.java b/....11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafka011Exception.java
@@ -17,11 +17,13 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.util.FlinkException;
 
 /**
  * Exception used by {@link FlinkKafkaProducer011} and {@link FlinkKafkaConsumer011}.
  */
+@PublicEvolving
 public class FlinkKafka011Exception extends FlinkException {
 
  private final FlinkKafka011ErrorCode errorCode;

diff --git a/...0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer011.java b/...0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer011.java
@@ -44,6 +44,7 @@
  * <p>Please refer to Kafka's documentation for the available configuration properties:
  * http:https://kafka.apache.org/documentation.html#newconsumerconfigs</p>
  */
+@PublicEvolving
 public class FlinkKafkaConsumer011<T> extends FlinkKafkaConsumer010<T> {
 
  private static final long serialVersionUID = 2324564345203409112L;

diff --git a/...11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSource.java b/...11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSource.java
@@ -18,6 +18,7 @@
 
 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.table.api.TableSchema;
 import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
@@ -35,6 +36,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.11.
  */
+@PublicEvolving
 public class Kafka011AvroTableSource extends KafkaAvroTableSource {
 
  /**

diff --git a/...11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSource.java b/...11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSource.java
@@ -18,6 +18,7 @@
 
 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.table.api.TableSchema;
 import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
@@ -32,6 +33,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.11.
  */
+@PublicEvolving
 public class Kafka011JsonTableSource extends KafkaJsonTableSource {
 
  /**

diff --git a/...a-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSource.java b/...a-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSource.java
@@ -18,6 +18,7 @@
 
 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.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.api.TableSchema;
@@ -29,6 +30,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.11.
  */
+@PublicEvolving
 public abstract class Kafka011TableSource extends KafkaTableSource {
 
  // The deserialization schema for the Kafka records

diff --git a/...rc/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java b/...rc/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java
@@ -18,6 +18,7 @@
 
 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;
 
@@ -101,6 +102,7 @@
  * required changes via Java Reflection API. It might not be the prettiest solution. An alternative would be to
  * re-implement whole Kafka's 0.11 REST API client on our own.
  */
+@PublicEvolving
 public class FlinkKafkaProducer<K, V> implements Producer<K, V> {
  private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaProducer.class);
 

diff --git a/.../java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java b/.../java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java
@@ -17,6 +17,8 @@
 
 package org.apache.flink.streaming.connectors.kafka.internal;
 
+import org.apache.flink.annotation.Internal;
+
 import java.util.HashSet;
 import java.util.Set;
 
@@ -34,6 +36,7 @@
  * </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;

diff --git a/.../apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMuttableWrapper.java b/.../apache/flink/streaming/connectors/kafka/internal/metrics/KafkaMetricMuttableWrapper.java
@@ -18,13 +18,15 @@
 
 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 KafkaMetricMuttableWrapper implements Gauge<Double> {
  private org.apache.kafka.common.Metric kafkaMetric;
 

diff --git a/...a-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java b/...a-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
@@ -81,6 +81,7 @@
  * <p>When using a Kafka topic to send data between Flink jobs, we recommend using the
  * {@see TypeInformationSerializationSchema} and {@see TypeInformationKeyValueSerializationSchema}.</p>
  */
+@PublicEvolving
 public class FlinkKafkaConsumer08<T> extends FlinkKafkaConsumerBase<T> {
 
  private static final long serialVersionUID = -6272159445203409112L;

diff --git a/...a-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer08.java b/...a-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer08.java
@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaDelegatePartitioner;
@@ -36,6 +37,7 @@
  *
  * @param <IN> Type of the messages to write into Kafka.
  */
+@PublicEvolving
 public class FlinkKafkaProducer08<IN> extends FlinkKafkaProducerBase<IN> {
 
  private static final long serialVersionUID = 1L;

diff --git a/...0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08AvroTableSource.java b/...0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08AvroTableSource.java
@@ -18,6 +18,7 @@
 
 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.table.api.TableSchema;
 import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
@@ -35,6 +36,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.8.
  */
+@PublicEvolving
 public class Kafka08AvroTableSource extends KafkaAvroTableSource {
 
  /**

diff --git a/...a-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java b/...a-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaDelegatePartitioner;
@@ -30,6 +31,7 @@
 /**
  * Kafka 0.8 {@link KafkaTableSink} that serializes data in JSON format.
  */
+@PublicEvolving
 public class Kafka08JsonTableSink extends KafkaJsonTableSink {
 
  /**

diff --git a/...0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSource.java b/...0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSource.java
@@ -18,6 +18,7 @@
 
 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.table.api.TableSchema;
 import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
@@ -32,6 +33,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.8.
  */
+@PublicEvolving
 public class Kafka08JsonTableSource extends KafkaJsonTableSource {
 
  /**

diff --git a/...fka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java b/...fka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java
@@ -18,6 +18,7 @@
 
 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.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.api.TableSchema;
@@ -29,6 +30,7 @@
 /**
  * Kafka {@link StreamTableSource} for Kafka 0.8.
  */
+@PublicEvolving
 public abstract class Kafka08TableSource extends KafkaTableSource {
 
  // The deserialization schema for the Kafka records

diff --git a/...8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/Kafka08Fetcher.java b/...8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/Kafka08Fetcher.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.connectors.kafka.internals;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
@@ -57,6 +58,7 @@
  *
  * @param <T> The type of elements produced by the fetcher.
  */
+@Internal
 public class Kafka08Fetcher<T> extends AbstractFetcher<T, TopicAndPartition> {
 
  static final KafkaTopicPartitionState<TopicAndPartition> MARKER =

diff --git a/...ava/org/apache/flink/streaming/connectors/kafka/internals/Kafka08PartitionDiscoverer.java b/...ava/org/apache/flink/streaming/connectors/kafka/internals/Kafka08PartitionDiscoverer.java
@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.connectors.kafka.internals;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.util.NetUtils;
 
 import kafka.cluster.Broker;
@@ -49,6 +50,7 @@
  * A partition discoverer that can be used to discover topics and partitions metadata
  * from Kafka brokers via the Kafka 0.8 low-level consumer API.
  */
+@Internal
 public class Kafka08PartitionDiscoverer extends AbstractPartitionDiscoverer {
 
  private static final Logger LOG = LoggerFactory.getLogger(Kafka08PartitionDiscoverer.class);