KAFKA-12579: Remove various deprecated clients classes/methods for 3.0 (apache#10438)

* Remove `ExtendedSerializer` and `ExtendedDeserializer`, deprecated since 2.1.
The extra functionality was also made available in `Serializer` and `Deserializer`.
* Remove `close(long, TimeUnit)` from the producer, consumer and admin client,
deprecated since 2.0 for the consumer and 2.2 for the rest. The replacement is `close(Duration)`.
* Remove `ConsumerConfig.addDeserializerToConfig` and `ProducerConfig.addSerializerToConfig`,
deprecated since 2.7 with no replacement. These methods were not intended to be public API
and are likely not used much (if at all).
* Remove `NoOffsetForPartitionException.partition()`, deprecated since 0.11. `partitions()`
should be used instead.
* Remove `MessageFormatter.init(Properties)`, deprecated since 2.7. The `configure(Map)`
method should be used instead.
* Remove `kafka.common.MessageFormatter`, deprecated since 2.7.
`org.apache.kafka.common.MessageFormatter` should be used instead.

Reviewers: Chia-Ping Tsai <[email protected]>, David Jacot <[email protected]>

Author: ijuma

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

@@ -24,7 +24,6 @@
 import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.ElectionType;
@@ -147,28 +146,11 @@ static Admin create(Map<String, Object> conf) {
     /**
      * Close the Admin and release all associated resources.
      * <p>
-     * See {@link #close(long, TimeUnit)}
+     * See {@link #close(Duration)}
      */
     @Override
     default void close() {
-        close(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
-    }
-
-    /**
-     * Close the Admin and release all associated resources.
-     * <p>
-     * The close operation has a grace period during which current operations will be allowed to
-     * complete, specified by the given duration and time unit.
-     * New operations will not be accepted during the grace period. Once the grace period is over,
-     * all operations that have not yet been completed will be aborted with a {@link org.apache.kafka.common.errors.TimeoutException}.
-     *
-     * @param duration The duration to use for the wait time.
-     * @param unit     The time unit to use for the wait time.
-     * @deprecated Since 2.2. Use {@link #close(Duration)} or {@link #close()}.
-     */
-    @Deprecated
-    default void close(long duration, TimeUnit unit) {
-        close(Duration.ofMillis(unit.toMillis(duration)));
+        close(Duration.ofMillis(Long.MAX_VALUE));
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java

@@ -28,7 +28,6 @@
 import java.util.Map;
 import java.util.OptionalLong;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 
 /**
@@ -264,12 +263,6 @@ public interface Consumer<K, V> extends Closeable {
      */
     void close();
 
-    /**
-     * @see KafkaConsumer#close(long, TimeUnit)
-     */
-    @Deprecated
-    void close(long timeout, TimeUnit unit);
-
     /**
      * @see KafkaConsumer#close(Duration)
      */

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -578,19 +578,9 @@ private void maybeOverrideClientId(Map<String, Object> configs) {
         }
     }
 
-    /**
-     * @deprecated Since 2.7.0. This will be removed in a future major release.
-     */
-    @Deprecated
-    public static Map<String, Object> addDeserializerToConfig(Map<String, Object> configs,
-                                                              Deserializer<?> keyDeserializer,
-                                                              Deserializer<?> valueDeserializer) {
-        return appendDeserializerToConfig(configs, keyDeserializer, valueDeserializer);
-    }
-
-    static Map<String, Object> appendDeserializerToConfig(Map<String, Object> configs,
-            Deserializer<?> keyDeserializer,
-            Deserializer<?> valueDeserializer) {
+    protected static Map<String, Object> appendDeserializerToConfig(Map<String, Object> configs,
+                                                                    Deserializer<?> keyDeserializer,
+                                                                    Deserializer<?> valueDeserializer) {
         Map<String, Object> newConfigs = new HashMap<>(configs);
         if (keyDeserializer != null)
             newConfigs.put(KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass());
@@ -599,22 +589,6 @@ static Map<String, Object> appendDeserializerToConfig(Map<String, Object> config
         return newConfigs;
     }
 
-    /**
-     * @deprecated Since 2.7.0. This will be removed in a future major release.
-     */
-    @Deprecated
-    public static Properties addDeserializerToConfig(Properties properties,
-                                                     Deserializer<?> keyDeserializer,
-                                                     Deserializer<?> valueDeserializer) {
-        Properties newProperties = new Properties();
-        newProperties.putAll(properties);
-        if (keyDeserializer != null)
-            newProperties.put(KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
-        if (valueDeserializer != null)
-            newProperties.put(VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
-        return newProperties;
-    }
-
     boolean maybeOverrideEnableAutoCommit() {
         Optional<String> groupId = Optional.ofNullable(getString(CommonClientConfigs.GROUP_ID_CONFIG));
         boolean enableAutoCommit = getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -2308,29 +2308,6 @@ public void close() {
         close(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS));
     }
 
-    /**
-     * Tries to close the consumer cleanly within the specified timeout. This method waits up to
-     * {@code timeout} for the consumer to complete pending commits and leave the group.
-     * If auto-commit is enabled, this will commit the current offsets if possible within the
-     * timeout. If the consumer is unable to complete offset commits and gracefully leave the group
-     * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
-     * used to interrupt close.
-     *
-     * @param timeout The maximum time to wait for consumer to close gracefully. The value must be
-     *                non-negative. Specifying a timeout of zero means do not wait for pending requests to complete.
-     * @param timeUnit The time unit for the {@code timeout}
-     * @throws IllegalArgumentException If the {@code timeout} is negative.
-     * @throws InterruptException If the thread is interrupted before or while this function is called
-     * @throws org.apache.kafka.common.KafkaException for any other error during close
-     *
-     * @deprecated Since 2.0. Use {@link #close(Duration)} or {@link #close()}.
-     */
-    @Deprecated
-    @Override
-    public void close(long timeout, TimeUnit timeUnit) {
-        close(Duration.ofMillis(timeUnit.toMillis(timeout)));
-    }
-
     /**
      * Tries to close the consumer cleanly within the specified timeout. This method waits up to
      * {@code timeout} for the consumer to complete pending commits and leave the group.

clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java

@@ -40,7 +40,6 @@
 import java.util.OptionalLong;
 import java.util.Queue;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -447,13 +446,7 @@ public synchronized Map<TopicPartition, Long> endOffsets(Collection<TopicPartiti
     }
 
     @Override
-    public synchronized void close() {
-        close(KafkaConsumer.DEFAULT_CLOSE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
-    }
-
-    @Deprecated
-    @Override
-    public synchronized void close(long timeout, TimeUnit unit) {
+    public final synchronized void close() {
         this.closed = true;
     }
 

clients/src/main/java/org/apache/kafka/clients/consumer/NoOffsetForPartitionException.java

@@ -43,16 +43,6 @@ public NoOffsetForPartitionException(Collection<TopicPartition> partitions) {
         this.partitions = Collections.unmodifiableSet(new HashSet<>(partitions));
     }
 
-    /**
-     * returns the first partition (out of {@link #partitions}) for which no offset is defined.
-     * @deprecated please use {@link #partitions}
-     * @return a partition with no offset
-     */
-    @Deprecated
-    public TopicPartition partition() {
-        return partitions.isEmpty() ? null : partitions.iterator().next();
-    }
-
     /**
      * returns all partitions for which no offests are defined.
      * @return all partitions without offsets

clients/src/main/java/org/apache/kafka/clients/producer/Producer.java

@@ -29,7 +29,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 
 /**
  * The interface for the {@link KafkaProducer}
@@ -100,11 +99,6 @@ void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
      */
     void close();
 
-    @Deprecated
-    default void close(long timeout, TimeUnit unit) {
-        close(Duration.ofMillis(unit.toMillis(timeout)));
-    }
-
     /**
      * See {@link KafkaProducer#close(Duration)}
      */

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -490,15 +490,6 @@ private static String parseAcks(String acksString) {
         }
     }
 
-    /**
-     * @deprecated Since 2.7.0. This will be removed in a future major release.
-     */
-    @Deprecated
-    public static Map<String, Object> addSerializerToConfig(Map<String, Object> configs,
-                                                            Serializer<?> keySerializer, Serializer<?> valueSerializer) {
-        return appendSerializerToConfig(configs, keySerializer, valueSerializer);
-    }
-
     static Map<String, Object> appendSerializerToConfig(Map<String, Object> configs,
             Serializer<?> keySerializer,
             Serializer<?> valueSerializer) {
@@ -510,22 +501,6 @@ static Map<String, Object> appendSerializerToConfig(Map<String, Object> configs,
         return newConfigs;
     }
 
-    /**
-     * @deprecated Since 2.7.0. This will be removed in a future major release.
-     */
-    @Deprecated
-    public static Properties addSerializerToConfig(Properties properties,
-                                                   Serializer<?> keySerializer,
-                                                   Serializer<?> valueSerializer) {
-        Properties newProperties = new Properties();
-        newProperties.putAll(properties);
-        if (keySerializer != null)
-            newProperties.put(KEY_SERIALIZER_CLASS_CONFIG, keySerializer.getClass().getName());
-        if (valueSerializer != null)
-            newProperties.put(VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer.getClass().getName());
-        return newProperties;
-    }
-
     public ProducerConfig(Properties props) {
         super(CONFIG, props);
     }

clients/src/main/java/org/apache/kafka/common/MessageFormatter.java

@@ -19,7 +19,6 @@
 import java.io.Closeable;
 import java.io.PrintStream;
 import java.util.Map;
-import java.util.Properties;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
@@ -34,33 +33,21 @@
  */
 public interface MessageFormatter extends Configurable, Closeable {
 
-    /**
-     * Initialises the MessageFormatter
-     * @param props Properties to configure the formatter
-     * @deprecated Use {@link #configure(Map)} instead, this method is for backward compatibility with the older Formatter interface
-     */
-    @Deprecated
-    default public void init(Properties props) {}
-
     /**
      * Configures the MessageFormatter
      * @param configs Map to configure the formatter
      */
-    default public void configure(Map<String, ?> configs) {
-        Properties properties = new Properties();
-        properties.putAll(configs);
-        init(properties);
-    }
+    default void configure(Map<String, ?> configs) {}
 
     /**
      * Parses and formats a record for display
      * @param consumerRecord the record to format
      * @param output the print stream used to output the record
      */
-    public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream output);
+    void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream output);
 
     /**
      * Closes the formatter
      */
-    default public void close() {}
+    default void close() {}
 }