[FLINK-36177] Deprecate KafkaShuffle and more

This commit deprecates all classes that are slated for removal in the kafka-4.0 release compatible with Flink 2.0.

I also deprecated internal classes to make later removal easier. Some public classes will cease to be public API but are still internally used.

Author: AHeise

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import org.apache.commons.collections.map.LinkedMap;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.ExecutionConfig;
@@ -58,8 +59,6 @@
 import org.apache.flink.streaming.runtime.operators.util.AssignerWithPunctuatedWatermarksAdapter;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.SerializedValue;
-
-import org.apache.commons.collections.map.LinkedMap;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -90,6 +89,7 @@
  * @param <T> The type of records produced by this data source
  */
 @Internal
+@Deprecated
 public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFunction<T>
         implements CheckpointListener, ResultTypeQueryable<T>, CheckpointedFunction {
 

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaErrorCode.java

@@ -19,8 +19,14 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 
-/** Error codes used in {@link FlinkKafkaException}. */
+/**
+ * Error codes used in {@link FlinkKafkaException}.
+ *
+ * @deprecated Will be removed with {@link FlinkKafkaProducer} and {@link
+ *     org.apache.flink.streaming.connectors.kafka.shuffle.FlinkKafkaShuffle}.
+ */
 @PublicEvolving
+@Deprecated
 public enum FlinkKafkaErrorCode {
     PRODUCERS_POOL_EMPTY,
     EXTERNAL_ERROR

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaException.java

@@ -20,8 +20,14 @@
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.util.FlinkException;
 
-/** Exception used by {@link FlinkKafkaProducer} and {@link FlinkKafkaConsumer}. */
+/**
+ * Exception used by {@link FlinkKafkaProducer} and {@link FlinkKafkaConsumer}.
+ *
+ * @deprecated Will be removed with {@link FlinkKafkaProducer} and {@link
+ *     org.apache.flink.streaming.connectors.kafka.shuffle.FlinkKafkaShuffle}.
+ */
 @PublicEvolving
+@Deprecated
 public class FlinkKafkaException extends FlinkException {
 
     private static final long serialVersionUID = 920269130311214200L;

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaContextAware.java

@@ -26,8 +26,11 @@
  *
  * <p>You only need to override the methods for the information that you need. However, {@link
  * #getTargetTopic(Object)} is required because it is used to determine the available partitions.
+ *
+ * @deprecated Will be turned into internal API when {@link FlinkKafkaProducer} is removed.
  */
 @PublicEvolving
+@Deprecated
 public interface KafkaContextAware<T> {
 
     /**

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaDeserializationSchema.java

@@ -21,7 +21,6 @@
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
 import org.apache.flink.util.Collector;
-
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 import java.io.Serializable;
@@ -31,8 +30,10 @@
  * (Java/Scala objects) that are processed by Flink.
  *
  * @param <T> The type created by the keyed deserialization schema.
+ * @deprecated Will be turned into internal API when {@link FlinkKafkaConsumer} is removed.
  */
 @PublicEvolving
+@Deprecated
 public interface KafkaDeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {
 
     /**

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java

@@ -19,11 +19,9 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
-
 import org.apache.kafka.clients.producer.ProducerRecord;
 
 import javax.annotation.Nullable;
-
 import java.io.Serializable;
 
 /**
@@ -35,8 +33,10 @@
  * which the Kafka Producer is running.
  *
  * @param <T> the type of values being serialized
+ * @deprecated Will be turned into internal API when {@link FlinkKafkaProducer} is removed.
  */
 @PublicEvolving
+@Deprecated
 public interface KafkaSerializationSchema<T> extends Serializable {
 
     /**

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/OffsetCommitMode.java

@@ -26,6 +26,7 @@
  * <p>The exact value of this is determined at runtime in the consumer subtasks.
  */
 @Internal
+@Deprecated
 public enum OffsetCommitMode {
 
     /** Completely disable offset committing. */

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/OffsetCommitModes.java

@@ -21,6 +21,7 @@
 
 /** Utilities for {@link OffsetCommitMode}. */
 @Internal
+@Deprecated
 public class OffsetCommitModes {
 
     /**

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java

@@ -32,7 +32,6 @@
 import org.apache.flink.util.SerializedValue;
 
 import javax.annotation.Nonnull;
-
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
@@ -61,6 +60,7 @@
  * @param <KPH> The type of topic/partition identifier used by Kafka in the specific version.
  */
 @Internal
+@Deprecated
 public abstract class AbstractFetcher<T, KPH> {
 
     private static final int NO_TIMESTAMPS_WATERMARKS = 0;

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractPartitionDiscoverer.java

@@ -41,6 +41,7 @@
  * allows the discoverer to be interrupted during a {@link #discoverPartitions()} call.
  */
 @Internal
+@Deprecated
 public abstract class AbstractPartitionDiscoverer {
 
     /** Describes whether we are discovering partitions for fixed topics or a topic pattern. */

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java

@@ -49,6 +49,7 @@
  * @param <E> The type of elements in the queue.
  */
 @Internal
+@Deprecated
 public class ClosableBlockingQueue<E> {
 
     /** The lock used to make queue accesses and open checks atomic. */

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ExceptionProxy.java

@@ -21,7 +21,6 @@
 import org.apache.flink.annotation.Internal;
 
 import javax.annotation.Nullable;
-
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -65,6 +64,7 @@
  * }</pre>
  */
 @Internal
+@Deprecated
 public class ExceptionProxy {
 
     /** The thread that should be interrupted when an exception occurs. */

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/FlinkKafkaInternalProducer.java

@@ -21,7 +21,6 @@
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.util.Preconditions;
-
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Callback;
@@ -43,7 +42,6 @@
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
-
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
@@ -58,6 +56,7 @@
 
 /** Internal flink kafka producer. */
 @PublicEvolving
+@Deprecated
 public class FlinkKafkaInternalProducer<K, V> implements Producer<K, V> {
     private static final Logger LOG = LoggerFactory.getLogger(FlinkKafkaInternalProducer.class);
 

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/Handover.java

@@ -20,12 +20,10 @@
 
 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;
@@ -47,6 +45,7 @@
  */
 @ThreadSafe
 @Internal
+@Deprecated
 public final class Handover implements Closeable {
 
     private final Object lock = new Object();

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaCommitCallback.java

@@ -25,6 +25,7 @@
  * commit request completes, which should normally be triggered from checkpoint complete event.
  */
 @Internal
+@Deprecated
 public interface KafkaCommitCallback {
 
     /**

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaConsumerThread.java

@@ -23,7 +23,6 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.metrics.MetricGroup;
 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;
@@ -35,7 +34,6 @@
 import org.slf4j.Logger;
 
 import javax.annotation.Nonnull;
-
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -61,6 +59,7 @@
  * an indirection to the KafkaConsumer calls that change signature.
  */
 @Internal
+@Deprecated
 public class KafkaConsumerThread<T> extends Thread {
 
     /** Logger for this consumer. */

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaDeserializationSchemaWrapper.java

@@ -22,7 +22,6 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
 import org.apache.flink.util.Collector;
-
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 /**
@@ -32,6 +31,7 @@
  * @param <T> The type created by the deserialization schema.
  */
 @Internal
+@Deprecated
 public class KafkaDeserializationSchemaWrapper<T> implements KafkaDeserializationSchema<T> {
 
     private static final long serialVersionUID = 2651665280744549932L;

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaFetcher.java

@@ -26,7 +26,6 @@
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.ExceptionUtils;
 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;
@@ -35,7 +34,6 @@
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
-
 import java.util.ArrayDeque;
 import java.util.HashMap;
 import java.util.List;
@@ -51,6 +49,7 @@
  * @param <T> The type of elements produced by the fetcher.
  */
 @Internal
+@Deprecated
 public class KafkaFetcher<T> extends AbstractFetcher<T, TopicPartition> {
 
     private static final Logger LOG = LoggerFactory.getLogger(KafkaFetcher.class);

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaPartitionDiscoverer.java

@@ -18,7 +18,6 @@
 package org.apache.flink.streaming.connectors.kafka.internals;
 
 import org.apache.flink.annotation.Internal;
-
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.PartitionInfo;
 
@@ -34,6 +33,7 @@
  * brokers via the Kafka high-level consumer API.
  */
 @Internal
+@Deprecated
 public class KafkaPartitionDiscoverer extends AbstractPartitionDiscoverer {
 
     private final Properties kafkaProperties;

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaSerializationSchemaWrapper.java

@@ -23,7 +23,6 @@
 import org.apache.flink.streaming.connectors.kafka.KafkaContextAware;
 import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-
 import org.apache.kafka.clients.producer.ProducerRecord;
 
 import javax.annotation.Nullable;
@@ -35,6 +34,7 @@
  * KafkaSerializationSchema}.
  */
 @Internal
+@Deprecated
 public class KafkaSerializationSchemaWrapper<T>
         implements KafkaSerializationSchema<T>, KafkaContextAware<T> {
 

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaShuffleFetcher.java

@@ -32,7 +32,6 @@
 import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.SerializedValue;
-
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
 
@@ -50,6 +49,7 @@
 
 /** Fetch data from Kafka for Kafka Shuffle. */
 @Internal
+@Deprecated
 public class KafkaShuffleFetcher<T> extends KafkaFetcher<T> {
     /** The handler to check and generate watermarks from fetched records. * */
     private final WatermarkHandler watermarkHandler;

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartition.java

@@ -32,8 +32,12 @@
  *
  * <p>Note: This class must not change in its structure, because it would change the serialization
  * format and make previous savepoints unreadable.
+ *
+ * @deprecated Will be turned into internal class when {@link
+ *     org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer} is removed.
  */
 @PublicEvolving
+@Deprecated
 public final class KafkaTopicPartition implements Serializable {
 
     /**

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionAssigner.java

@@ -21,6 +21,7 @@
 
 /** Utility for assigning Kafka partitions to consumer subtasks. */
 @Internal
+@Deprecated
 public class KafkaTopicPartitionAssigner {
 
     /**

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionLeader.java

@@ -18,7 +18,6 @@
 package org.apache.flink.streaming.connectors.kafka.internals;
 
 import org.apache.flink.annotation.Internal;
-
 import org.apache.kafka.common.Node;
 
 import java.io.Serializable;
@@ -27,6 +26,7 @@
  * Serializable Topic Partition info with leader Node information. This class is used at runtime.
  */
 @Internal
+@Deprecated
 public class KafkaTopicPartitionLeader implements Serializable {
 
     private static final long serialVersionUID = 9145855900303748582L;

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionState.java

@@ -29,6 +29,7 @@
  * @param <KPH> The type of the Kafka partition descriptor, which varies across Kafka versions.
  */
 @Internal
+@Deprecated
 public class KafkaTopicPartitionState<T, KPH> {
 
     // ------------------------------------------------------------------------

Diff for: flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateSentinel.java

@@ -26,6 +26,7 @@
  * number that is probably (hopefully) not used by Kafka as a magic number for anything else.
  */
 @Internal
+@Deprecated
 public class KafkaTopicPartitionStateSentinel {
 
     /** Magic number that defines an unset offset. */