[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

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaDeserializationSchemaWrapper.java

@@ -33,7 +33,9 @@
  * ConsumerRecord ConsumerRecords}.
  *
  * @param <T> the type of the deserialized records.
+ * @deprecated Remove with @{@link KafkaDeserializationSchema}
  */
+@Deprecated
 class KafkaDeserializationSchemaWrapper<T> implements KafkaRecordDeserializationSchema<T> {
     private static final long serialVersionUID = 1L;
     private final KafkaDeserializationSchema<T> kafkaDeserializationSchema;

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java

@@ -71,7 +71,9 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
      * @param <V> the return type of the deserialized record.
      * @return A {@link KafkaRecordDeserializationSchema} that uses the given {@link
      *     KafkaDeserializationSchema} to deserialize the {@link ConsumerRecord ConsumerRecords}.
+     * @deprecated Will be removed with {@link KafkaDeserializationSchema}.
      */
+    @Deprecated
     static <V> KafkaRecordDeserializationSchema<V> of(
             KafkaDeserializationSchema<V> kafkaDeserializationSchema) {
         return new KafkaDeserializationSchemaWrapper<>(kafkaDeserializationSchema);

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

@@ -90,6 +90,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 {
 

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

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;

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> {
 
     /**

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

@@ -31,8 +31,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> {
 
     /**

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

@@ -35,8 +35,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 {
 
     /**

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. */

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 {
 
     /**

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

@@ -61,6 +61,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;

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. */

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. */

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

@@ -65,6 +65,7 @@
  * }</pre>
  */
 @Internal
+@Deprecated
 public class ExceptionProxy {
 
     /** The thread that should be interrupted when an exception occurs. */

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

@@ -58,6 +58,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);
 

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

@@ -47,6 +47,7 @@
  */
 @ThreadSafe
 @Internal
+@Deprecated
 public final class Handover implements Closeable {
 
     private final Object lock = new Object();

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 {
 
     /**

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

@@ -61,6 +61,7 @@
  * an indirection to the KafkaConsumer calls that change signature.
  */
 @Internal
+@Deprecated
 public class KafkaConsumerThread<T> extends Thread {
 
     /** Logger for this consumer. */

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

@@ -32,6 +32,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;

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

@@ -51,6 +51,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);

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

@@ -34,6 +34,7 @@
  * brokers via the Kafka high-level consumer API.
  */
 @Internal
+@Deprecated
 public class KafkaPartitionDiscoverer extends AbstractPartitionDiscoverer {
 
     private final Properties kafkaProperties;

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

@@ -35,6 +35,7 @@
  * KafkaSerializationSchema}.
  */
 @Internal
+@Deprecated
 public class KafkaSerializationSchemaWrapper<T>
         implements KafkaSerializationSchema<T>, KafkaContextAware<T> {
 

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

@@ -50,6 +50,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;

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

@@ -32,8 +32,13 @@
  *
  * <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. Replace with
+ *     {@link org.apache.kafka.common.TopicPartition}.
  */
 @PublicEvolving
+@Deprecated
 public final class KafkaTopicPartition implements Serializable {
 
     /**

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 {
 
     /**

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

@@ -27,6 +27,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;

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> {
 
     // ------------------------------------------------------------------------

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithWatermarkGenerator.java

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

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaTopicsDescriptor.java

@@ -33,6 +33,7 @@
  * list of topics, or a topic pattern.
  */
 @Internal
+@Deprecated
 public class KafkaTopicsDescriptor implements Serializable {
 
     private static final long serialVersionUID = -3807227764764900975L;

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KeyedSerializationSchemaWrapper.java

@@ -27,6 +27,7 @@
  * @param <T> The type to serialize
  */
 @Internal
+@Deprecated
 public class KeyedSerializationSchemaWrapper<T> implements KeyedSerializationSchema<T> {
 
     private static final long serialVersionUID = 1351665280744549933L;

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SourceContextWatermarkOutputAdapter.java

@@ -25,6 +25,7 @@
  * A {@link org.apache.flink.api.common.eventtime.WatermarkOutput} that forwards calls to a {@link
  * org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext}.
  */
+@Deprecated
 public class SourceContextWatermarkOutputAdapter<T> implements WatermarkOutput {
     private final SourceContext<T> sourceContext;
 

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/TransactionalIdsGenerator.java

@@ -40,6 +40,7 @@
  * subtask.
  */
 @Internal
+@Deprecated
 public class TransactionalIdsGenerator {
     private final String prefix;
     private final int subtaskIndex;

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/metrics/KafkaConsumerMetricConstants.java

@@ -26,6 +26,7 @@
  * metrics.
  */
 @Internal
+@Deprecated
 public class KafkaConsumerMetricConstants {
 
     public static final String KAFKA_CONSUMER_METRICS_GROUP = "KafkaConsumer";

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/metrics/KafkaMetricWrapper.java

@@ -23,6 +23,7 @@
 
 /** Gauge for getting the current value of a Kafka metric. */
 @Internal
+@Deprecated
 public class KafkaMetricWrapper implements Gauge<Double> {
     private final org.apache.kafka.common.Metric kafkaMetric;
 

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/FlinkFixedPartitioner.java

@@ -54,8 +54,12 @@
  * <p>Not all Kafka partitions contain data To avoid such an unbalanced partitioning, use a
  * round-robin kafka partitioner (note that this will cause a lot of network connections between all
  * the Flink instances and all the Kafka brokers).
+ *
+ * @deprecated Will be turned into internal class when {@link
+ *     org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer} is removed.
  */
 @PublicEvolving
+@Deprecated
 public class FlinkFixedPartitioner<T> extends FlinkKafkaPartitioner<T> {
 
     private static final long serialVersionUID = -3785320239953858777L;

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/FlinkKafkaPartitioner.java

@@ -24,8 +24,12 @@
 /**
  * A {@link FlinkKafkaPartitioner} wraps logic on how to partition records across partitions of
  * multiple Kafka topics.
+ *
+ * @deprecated Will be turned into internal class when {@link
+ *     org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer} is removed.
  */
 @PublicEvolving
+@Deprecated
 public abstract class FlinkKafkaPartitioner<T> implements Serializable {
 
     private static final long serialVersionUID = -9086719227828020494L;

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/shuffle/FlinkKafkaShuffle.java

@@ -98,8 +98,13 @@
  *                                                |
  *                                                | ----------> KafkaShuffleConsumerReuse -> ...
  * </pre>
+ *
+ * @deprecated This experimental feature never graduated to a stable feature and will be removed in
+ *     future releases. In case of interest to port it to the Source/Sink API, please reach out to
+ *     the Flink community.
  */
 @Experimental
+@Deprecated
 public class FlinkKafkaShuffle {
     static final String PRODUCER_PARALLELISM = "producer parallelism";
     static final String PARTITION_NUMBER = "partition number";

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/shuffle/FlinkKafkaShuffleConsumer.java

@@ -39,6 +39,7 @@
 
 /** Flink Kafka Shuffle Consumer Function. */
 @Internal
+@Deprecated
 public class FlinkKafkaShuffleConsumer<T> extends FlinkKafkaConsumer<T> {
     private final TypeSerializer<T> typeSerializer;
     private final int producerParallelism;

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/shuffle/FlinkKafkaShuffleProducer.java

@@ -44,6 +44,7 @@
  * handling elements and watermarks
  */
 @Internal
+@Deprecated
 public class FlinkKafkaShuffleProducer<IN, KEY> extends FlinkKafkaProducer<IN> {
     private final KafkaSerializer<IN> kafkaSerializer;
     private final KeySelector<IN, KEY> keySelector;

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/shuffle/StreamKafkaShuffleSink.java

@@ -29,6 +29,7 @@
  * this way to avoid public interface change.
  */
 @Internal
+@Deprecated
 class StreamKafkaShuffleSink<IN> extends StreamSink<IN> {
 
     public StreamKafkaShuffleSink(FlinkKafkaShuffleProducer flinkKafkaShuffleProducer) {

flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaDeserializationSchema.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.connectors.kafka.table;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
@@ -38,6 +39,7 @@
 import java.util.List;
 
 /** A specific {@link KafkaSerializationSchema} for {@link KafkaDynamicSource}. */
+@Internal
 class DynamicKafkaDeserializationSchema implements KafkaDeserializationSchema<RowData> {
 
     private static final long serialVersionUID = 1L;