MINOR: Specify 2.1 as the minimum broker version in KafkaConsumer javadoc

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

@@ -117,7 +117,7 @@ import java.util.Set;
  *
  * <h3>Broker Compatibility</h3>
  * <p>
- * The minimum broker version required is 0.10.0.0. Methods with stricter requirements will specify the minimum broker
+ * The minimum broker version required is 2.1. Methods with stricter requirements will specify the minimum broker
  * version required.
  * <p>
  */
@@ -170,8 +170,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * This is a convenience method for {@link #createTopics(Collection, CreateTopicsOptions)} with default options.
      * See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.10.1.0 or higher.
      *
      * @param newTopics The new topics to create.
      * @return The CreateTopicsResult.
@@ -189,9 +187,6 @@ public interface Admin extends AutoCloseable {
      * success for all the brokers to become aware that the topics have been created.
      * During this time, {@link #listTopics()} and {@link #describeTopics(Collection)}
      * may not return information about the new topics.
-     * <p>
-     * This operation is supported by brokers with version 0.10.1.0 or higher. The validateOnly option is supported
-     * from version 0.10.2.0.
      *
      * @param newTopics The new topics to create.
      * @param options   The options to use when creating the new topics.
@@ -202,8 +197,6 @@ public interface Admin extends AutoCloseable {
     /**
      * This is a convenience method for {@link #deleteTopics(TopicCollection, DeleteTopicsOptions)}
      * with default options. See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.10.1.0 or higher.
      *
      * @param topics The topic names to delete.
      * @return The DeleteTopicsResult.
@@ -215,8 +208,6 @@ public interface Admin extends AutoCloseable {
     /**
      * This is a convenience method for {@link #deleteTopics(TopicCollection, DeleteTopicsOptions)}
      * with default options. See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.10.1.0 or higher.
      *
      * @param topics  The topic names to delete.
      * @param options The options to use when deleting the topics.
@@ -231,7 +222,6 @@ public interface Admin extends AutoCloseable {
      * with default options. See the overload for more details.
      * <p>
      * When using topic IDs, this operation is supported by brokers with inter-broker protocol 2.8 or higher.
-     * When using topic names, this operation is supported by brokers with version 0.10.1.0 or higher.
      *
      * @param topics The topics to delete.
      * @return The DeleteTopicsResult.
@@ -255,7 +245,6 @@ public interface Admin extends AutoCloseable {
      * return successfully in this case.
      * <p>
      * When using topic IDs, this operation is supported by brokers with inter-broker protocol 2.8 or higher.
-     * When using topic names, this operation is supported by brokers with version 0.10.1.0 or higher.
      *
      * @param topics  The topics to delete.
      * @param options The options to use when deleting the topics.
@@ -354,8 +343,6 @@ public interface Admin extends AutoCloseable {
     /**
      * This is a convenience method for {@link #describeAcls(AclBindingFilter, DescribeAclsOptions)} with
      * default options. See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param filter The filter to use.
      * @return The DescribeAclsResult.
@@ -369,8 +356,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * Note: it may take some time for changes made by {@code createAcls} or {@code deleteAcls} to be reflected
      * in the output of {@code describeAcls}.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param filter  The filter to use.
      * @param options The options to use when listing the ACLs.
@@ -381,8 +366,6 @@ public interface Admin extends AutoCloseable {
     /**
      * This is a convenience method for {@link #createAcls(Collection, CreateAclsOptions)} with
      * default options. See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param acls The ACLs to create
      * @return The CreateAclsResult.
@@ -398,8 +381,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * If you attempt to add an ACL that duplicates an existing ACL, no error will be raised, but
      * no changes will be made.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param acls    The ACLs to create
      * @param options The options to use when creating the ACLs.
@@ -410,8 +391,6 @@ public interface Admin extends AutoCloseable {
     /**
      * This is a convenience method for {@link #deleteAcls(Collection, DeleteAclsOptions)} with default options.
      * See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param filters The filters to use.
      * @return The DeleteAclsResult.
@@ -424,8 +403,6 @@ public interface Admin extends AutoCloseable {
      * Deletes access control lists (ACLs) according to the supplied filters.
      * <p>
      * This operation is not transactional so it may succeed for some ACLs while fail for others.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param filters The filters to use.
      * @param options The options to use when deleting the ACLs.
@@ -439,8 +416,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * This is a convenience method for {@link #describeConfigs(Collection, DescribeConfigsOptions)} with default options.
      * See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param resources See relevant type {@link ConfigResource.Type}
      * @return The DescribeConfigsResult
@@ -472,8 +447,6 @@ public interface Admin extends AutoCloseable {
      *     will throw a {@link org.apache.kafka.common.errors.TimeoutException} exception</li>
      *     <li>{@link ConfigResource.Type#CLIENT_METRICS}: will return empty configs</li>
      * </ul>
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param resources See relevant type {@link ConfigResource.Type}
      * @param options   The options to use when describing configs
@@ -534,8 +507,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * This is a convenience method for {@link #alterReplicaLogDirs(Map, AlterReplicaLogDirsOptions)} with default options.
      * See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 1.1.0 or higher.
      *
      * @param replicaAssignment     The replicas with their log directory absolute path
      * @return                      The AlterReplicaLogDirsResult
@@ -551,8 +522,6 @@ public interface Admin extends AutoCloseable {
      * log directory if it is not already there. For detailed result, inspect the returned {@link AlterReplicaLogDirsResult} instance.
      * <p>
      * This operation is not transactional so it may succeed for some replicas while fail for others.
-     * <p>
-     * This operation is supported by brokers with version 1.1.0 or higher.
      *
      * @param replicaAssignment     The replicas with their log directory absolute path
      * @param options               The options to use when changing replica dir
@@ -566,8 +535,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * This is a convenience method for {@link #describeLogDirs(Collection, DescribeLogDirsOptions)} with default options.
      * See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 1.0.0 or higher.
      *
      * @param brokers A list of brokers
      * @return The DescribeLogDirsResult
@@ -578,8 +545,6 @@ public interface Admin extends AutoCloseable {
 
     /**
      * Query the information of all log directories on the given set of brokers
-     * <p>
-     * This operation is supported by brokers with version 1.0.0 or higher.
      *
      * @param brokers A list of brokers
      * @param options The options to use when querying log dir info
@@ -592,8 +557,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * This is a convenience method for {@link #describeReplicaLogDirs(Collection, DescribeReplicaLogDirsOptions)}
      * with default options. See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 1.0.0 or higher.
      *
      * @param replicas The replicas to query
      * @return The DescribeReplicaLogDirsResult
@@ -604,8 +567,6 @@ public interface Admin extends AutoCloseable {
 
     /**
      * Query the replica log directory information for the specified replicas.
-     * <p>
-     * This operation is supported by brokers with version 1.0.0 or higher.
      *
      * @param replicas The replicas to query
      * @param options  The options to use when querying replica log dir info
@@ -641,8 +602,6 @@ public interface Admin extends AutoCloseable {
      * During this time, {@link #describeTopics(Collection)}
      * may not return information about the new partitions.
      * <p>
-     * This operation is supported by brokers with version 1.0.0 or higher.
-     * <p>
      * The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
      * {@link CreatePartitionsResult#values() values()} method of the returned {@link CreatePartitionsResult}
      * <ul>
@@ -674,8 +633,6 @@ public interface Admin extends AutoCloseable {
      * <p>
      * This is a convenience method for {@link #deleteRecords(Map, DeleteRecordsOptions)} with default options.
      * See the overload for more details.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param recordsToDelete The topic partitions and related offsets from which records deletion starts.
      * @return The DeleteRecordsResult.
@@ -686,8 +643,6 @@ public interface Admin extends AutoCloseable {
 
     /**
      * Delete records whose offset is smaller than the given offset of the corresponding partition.
-     * <p>
-     * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param recordsToDelete The topic partitions and related offsets from which records deletion starts.
      * @param options         The options to use when deleting records.
@@ -712,8 +667,6 @@ public interface Admin extends AutoCloseable {
     /**
      * Create a Delegation Token.
      * <p>
-     * This operation is supported by brokers with version 1.1.0 or higher.
-     * <p>
      * The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
      * {@link CreateDelegationTokenResult#delegationToken() delegationToken()} method of the returned {@link CreateDelegationTokenResult}
      * <ul>
@@ -749,8 +702,6 @@ public interface Admin extends AutoCloseable {
     /**
      * Renew a Delegation Token.
      * <p>
-     * This operation is supported by brokers with version 1.1.0 or higher.
-     * <p>
      * The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
      * {@link RenewDelegationTokenResult#expiryTimestamp() expiryTimestamp()} method of the returned {@link RenewDelegationTokenResult}
      * <ul>
@@ -790,8 +741,6 @@ public interface Admin extends AutoCloseable {
     /**
      * Expire a Delegation Token.
      * <p>
-     * This operation is supported by brokers with version 1.1.0 or higher.
-     * <p>
      * The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
      * {@link ExpireDelegationTokenResult#expiryTimestamp() expiryTimestamp()} method of the returned {@link ExpireDelegationTokenResult}
      * <ul>
@@ -830,8 +779,6 @@ public interface Admin extends AutoCloseable {
     /**
      * Describe the Delegation Tokens.
      * <p>
-     * This operation is supported by brokers with version 1.1.0 or higher.
-     * <p>
      * The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the
      * {@link DescribeDelegationTokenResult#delegationTokens() delegationTokens()} method of the returned {@link DescribeDelegationTokenResult}
      * <ul>
@@ -1094,7 +1041,7 @@ public interface Admin extends AutoCloseable {
      * {@link #describeTopics(Collection)} may not return information about the partitions'
      * new leaders.
      * <p>
-     * This operation is supported by brokers with version 2.2.0 or later if preferred election is use;
+     * This operation is supported by brokers with version 2.2.0 or later if preferred election is used;
      * otherwise the brokers most be 2.4.0 or higher.
      * <p>
      * The following exceptions can be anticipated when calling {@code get()} on the future obtained

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

@@ -61,9 +61,9 @@ import static org.apache.kafka.common.utils.Utils.propsToMap;
  * The consumer is not thread-safe. See <a href="#multithreaded">Multi-threaded Processing</a> for more details.
  *
  * <h3>Cross-Version Compatibility</h3>
- * This client can communicate with brokers that are version 0.10.0 or newer. Older or newer brokers may not support
+ * This client can communicate with brokers that are version 2.1 or newer. Older or newer brokers may not support
- * certain features. For example, 0.10.0 brokers do not support offsetsForTimes, because this feature was added
+ * certain features. For example, 3.9 brokers do not support {@code subscribe(SubscriptionPattern)}, because this feature was added
- * in version 0.10.1. You will receive an {@link org.apache.kafka.common.errors.UnsupportedVersionException}
+ * in version 4.0. You will receive an {@link org.apache.kafka.common.errors.UnsupportedVersionException}
  * when invoking an API that is not available on the running broker version.
  * <p>
  *

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

@@ -138,14 +138,14 @@ import java.util.concurrent.atomic.AtomicReference;
  * the <code>batch.size</code> config. Making this larger can result in more batching, but requires more memory (since we will
  * generally have one of these buffers for each active partition).
  * <p>
- * By default a buffer is available to send immediately even if there is additional unused space in the buffer. However if you
+ * By default, a buffer is available to send after a short delay even if there is additional unused space in the buffer.
- * want to reduce the number of requests you can set <code>linger.ms</code> to something greater than 0. This will
+ * However, if you want to reduce the number of requests you can set <code>linger.ms</code> to something greater than {@code 5}.
- * instruct the producer to wait up to that number of milliseconds before sending a request in hope that more records will
+ * This will instruct the producer to wait up to that number of milliseconds before sending a request in hope that more records will
  * arrive to fill up the same batch. This is analogous to Nagle's algorithm in TCP. For example, in the code snippet above,
- * likely all 100 records would be sent in a single request since we set our linger time to 1 millisecond. However this setting
+ * likely all 100 records would be sent in a single request since we set our linger time to 1 millisecond. This setting
- * would add 1 millisecond of latency to our request waiting for more records to arrive if we didn't fill up the buffer. Note that
+ * would also add 1 millisecond of latency to our request waiting for more records to arrive if we didn't fill up the buffer. Note that
  * records that arrive close together in time will generally batch together even with <code>linger.ms=0</code>. So, under heavy load,
- * batching will occur regardless of the linger configuration; however setting this to something larger than 0 can lead to fewer, more
+ * batching will occur regardless of the linger configuration; however setting this to something larger can lead to fewer, more
  * efficient requests when not under maximal load at the cost of a small amount of latency.
  * <p>
  * The <code>buffer.memory</code> controls the total amount of memory available to the producer for buffering. If records
@@ -157,24 +157,19 @@ import java.util.concurrent.atomic.AtomicReference;
  * their <code>ProducerRecord</code> into bytes. You can use the included {@link org.apache.kafka.common.serialization.ByteArraySerializer} or
  * {@link org.apache.kafka.common.serialization.StringSerializer} for simple byte or string types.
  * <p>
- * From Kafka 0.11, the KafkaProducer supports two additional modes: the idempotent producer and the transactional producer.
+ * The KafkaProducer supports two additional modes: the idempotent producer (enabled by default) and the transactional producer.
  * The idempotent producer strengthens Kafka's delivery semantics from at least once to exactly once delivery. In particular
  * producer retries will no longer introduce duplicates. The transactional producer allows an application to send messages
  * to multiple partitions (and topics!) atomically.
  * </p>
  * <p>
- * From Kafka 3.0, the <code>enable.idempotence</code> configuration defaults to true. When enabling idempotence,
+ * To ensure idempotence, it is imperative to avoid application level re-sends since these cannot be de-duplicated.
- * <code>retries</code> config will default to <code>Integer.MAX_VALUE</code> and the <code>acks</code> config will
+ * To achieve this, it is recommended to set {@code delivery.timeout.ms} such that retries are handled for the desired
- * default to <code>all</code>. There are no API changes for the idempotent producer, so existing applications will
+ * duration by the producer (the {@code retries} config should be left unset - the default is {@code Integer.MAX_VALUE}).
- * not need to be modified to take advantage of this feature.
+ * Additionally, if a {@link #send(ProducerRecord)} returns an error even with infinite retries (for instance if the
- * </p>
+ * message expires in the buffer before being sent), then it is recommended to shut down the producer and check the
- * <p>
+ * contents of the last produced message to ensure that it is not duplicated. Finally, the producer can only guarantee
- * To take advantage of the idempotent producer, it is imperative to avoid application level re-sends since these cannot
+ * idempotence for messages sent within a single session.
- * be de-duplicated. As such, if an application enables idempotence, it is recommended to leave the <code>retries</code>
- * config unset, as it will be defaulted to <code>Integer.MAX_VALUE</code>. Additionally, if a {@link #send(ProducerRecord)}
- * returns an error even with infinite retries (for instance if the message expires in the buffer before being sent),
- * then it is recommended to shut down the producer and check the contents of the last produced message to ensure that
- * it is not duplicated. Finally, the producer can only guarantee idempotence for messages sent within a single session.
  * </p>
  * <p>To use the transactional producer and the attendant APIs, you must set the <code>transactional.id</code>
  * configuration property. If the <code>transactional.id</code> is set, idempotence is automatically enabled along with
@@ -234,9 +229,10 @@ import java.util.concurrent.atomic.AtomicReference;
  * successful writes are marked as aborted, hence keeping the transactional guarantees.
  * </p>
  * <p>
- * This client can communicate with brokers that are version 0.10.0 or newer. Older or newer brokers may not support
+ * This client can communicate with brokers that are version 2.1 or newer. Older brokers may not support
- * certain client features.  For instance, the transactional APIs need broker versions 0.11.0 or later. You will receive an
+ * certain client features. For instance, {@code sendOffsetsToTransaction} with all consumer group metadata needs broker
- * <code>UnsupportedVersionException</code> when invoking an API that is not available in the running broker version.
+ * versions 2.5 or later. You will receive an <code>UnsupportedVersionException</code> when invoking an API that is not
+ * available in the running broker version.
  * </p>
  */
 public class KafkaProducer<K, V> implements Producer<K, V> {
@@ -636,8 +632,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * initialized, this method should no longer be used.
      *
      * @throws IllegalStateException if no {@code transactional.id} has been configured
-     * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
-     *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException error indicating that the configured
      *         transactional.id is not authorized, or the idempotent producer id is unavailable. See the exception for
      *         more details.  User may retry this function call after fixing the permission.
@@ -665,8 +659,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws ProducerFencedException if another producer with the same transactional.id is active
      * @throws org.apache.kafka.common.errors.InvalidProducerEpochException if the producer has attempted to produce with an old epoch
      *         to the partition leader. See the exception for more details
-     * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
-     *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
      *         {@code transactional.id} is not authorized. See the exception for more details
      * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error
@@ -693,7 +685,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * {@link KafkaConsumer#groupMetadata()} to leverage consumer group metadata. This will provide
      * stronger fencing than just supplying the {@code consumerGroupId} and passing in {@code new ConsumerGroupMetadata(consumerGroupId)},
      * however note that the full set of consumer group metadata returned by {@link KafkaConsumer#groupMetadata()}
-     * requires the brokers to be on version 2.5 or newer to understand.
+     * requires the brokers to be on version 2.5 or newer.
      *
      * <p>
      * This method is a blocking call that waits until the request has been received and acknowledged by the consumer group
@@ -710,8 +702,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws IllegalStateException if no transactional.id has been configured or no transaction has been started.
      * @throws ProducerFencedException fatal error indicating another producer with the same transactional.id is active
      * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
-     *         does not support transactions (i.e. if its version is lower than 0.11.0.0) or
+     *         doesn't support the latest version of transactional API with all consumer group metadata
-     *         the broker doesn't support the latest version of transactional API with all consumer group metadata
      *         (i.e. if its version is lower than 2.5.0).
      * @throws org.apache.kafka.common.errors.UnsupportedForMessageFormatException fatal error indicating the message
      *         format used for the offsets topic on the broker does not support transactions
@@ -764,8 +755,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      *
      * @throws IllegalStateException if no transactional.id has been configured or no transaction has been started
      * @throws ProducerFencedException fatal error indicating another producer with the same transactional.id is active
-     * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
-     *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
      *         transactional.id is not authorized. See the exception for more details
      * @throws org.apache.kafka.common.errors.InvalidProducerEpochException if the producer has attempted to produce with an old epoch
@@ -801,8 +790,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws ProducerFencedException fatal error indicating another producer with the same transactional.id is active
      * @throws org.apache.kafka.common.errors.InvalidProducerEpochException if the producer has attempted to produce with an old epoch
      *         to the partition leader. See the exception for more details
-     * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
-     *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
      *         transactional.id is not authorized. See the exception for more details
      * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error