MINOR: remove unneccessary public keyword from ProducerInterceptor/ConsumerInterceptor interface

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

@@ -338,7 +338,7 @@ default DescribeClusterResult describeCluster() {
      * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param filter The filter to use.
-     * @return The DeleteAclsResult.
+     * @return The DescribeAclsResult.
      */
     default DescribeAclsResult describeAcls(AclBindingFilter filter) {
         return describeAcls(filter, new DescribeAclsOptions());
@@ -354,7 +354,7 @@ default DescribeAclsResult describeAcls(AclBindingFilter filter) {
      *
      * @param filter  The filter to use.
      * @param options The options to use when listing the ACLs.
-     * @return The DeleteAclsResult.
+     * @return The DescribeAclsResult.
      */
     DescribeAclsResult describeAcls(AclBindingFilter filter, DescribeAclsOptions options);
 

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

@@ -65,7 +65,7 @@ public interface ConsumerInterceptor<K, V> extends Configurable, AutoCloseable {
      * @param records records to be consumed by the client or records returned by the previous interceptors in the list.
      * @return records that are either modified by the interceptor or same as records passed to this method.
      */
-    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
+    ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
 
     /**
      * This is called when offsets get committed.
@@ -74,10 +74,10 @@ public interface ConsumerInterceptor<K, V> extends Configurable, AutoCloseable {
      *
      * @param offsets A map of offsets by partition with associated metadata
      */
-    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
+    void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
 
     /**
      * This is called when interceptor is closed
      */
-    public void close();
+    void close();
 }

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

@@ -1901,7 +1901,7 @@ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition
      *
      * @param topic The topic to get partition metadata for
      *
-     * @return The list of partitions
+     * @return The list of partitions, which will be empty when the given topic is not found
      * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
@@ -1924,7 +1924,7 @@ public List<PartitionInfo> partitionsFor(String topic) {
      * @param topic The topic to get partition metadata for
      * @param timeout The maximum of time to await topic metadata
      *
-     * @return The list of partitions
+     * @return The list of partitions, which will be empty when the given topic is not found
      * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
@@ -1948,7 +1948,7 @@ public List<PartitionInfo> partitionsFor(String topic, Duration timeout) {
             Timer timer = time.timer(timeout);
             Map<String, List<PartitionInfo>> topicMetadata = fetcher.getTopicMetadata(
                     new MetadataRequest.Builder(Collections.singletonList(topic), metadata.allowAutoTopicCreation()), timer);
-            return topicMetadata.get(topic);
+            return topicMetadata.getOrDefault(topic, Collections.emptyList());
         } finally {
             release();
         }

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

@@ -376,7 +376,7 @@ public synchronized void updateEndOffsets(final Map<TopicPartition, Long> newOff
     @Override
     public synchronized List<PartitionInfo> partitionsFor(String topic) {
         ensureNotClosed();
-        return this.partitions.get(topic);
+        return this.partitions.getOrDefault(topic, Collections.emptyList());
     }
 
     @Override

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

@@ -200,8 +200,8 @@ public class ProducerConfig extends AbstractConfig {
     /** <code>max.in.flight.requests.per.connection</code> */
     public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection";
     private static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC = "The maximum number of unacknowledged requests the client will send on a single connection before blocking."
-                                                                            + " Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of"
-                                                                            + " message re-ordering due to retries (i.e., if retries are enabled).";
+                                                                            + " Note that if this config is set to be greater than 1 and <code>enable.idempotence</code> is set to false, there is a risk of"
+                                                                            + " message re-ordering after a failed send due to retries (i.e., if retries are enabled).";
 
     /** <code>retries</code> */
     public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG;
@@ -246,10 +246,10 @@ public class ProducerConfig extends AbstractConfig {
     public static final String ENABLE_IDEMPOTENCE_CONFIG = "enable.idempotence";
     public static final String ENABLE_IDEMPOTENCE_DOC = "When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream. If 'false', producer "
                                                         + "retries due to broker failures, etc., may write duplicates of the retried message in the stream. "
-                                                        + "Note that enabling idempotence requires <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to be less than or equal to 5, "
-                                                        + "<code>" + RETRIES_CONFIG + "</code> to be greater than 0 and <code>" + ACKS_CONFIG + "</code> must be 'all'. If these values "
-                                                        + "are not explicitly set by the user, suitable values will be chosen. If incompatible values are set, "
-                                                        + "a <code>ConfigException</code> will be thrown.";
+                                                        + "Note that enabling idempotence requires <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> to be less than or equal to 5 "
+                                                        + "(with message ordering preserved for any allowable value), <code>" + RETRIES_CONFIG + "</code> to be greater than 0, and <code>"
+                                                        + ACKS_CONFIG + "</code> must be 'all'. If these values are not explicitly set by the user, suitable values will be chosen. If incompatible "
+                                                        + "values are set, a <code>ConfigException</code> will be thrown.";
 
     /** <code> transaction.timeout.ms </code> */
     public static final String TRANSACTION_TIMEOUT_CONFIG = "transaction.timeout.ms";

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

@@ -64,7 +64,7 @@ public interface ProducerInterceptor<K, V> extends Configurable {
      * @param record the record from client or the record returned by the previous interceptor in the chain of interceptors.
      * @return producer record to send to topic/partition
      */
-    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
+    ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
 
     /**
      * This method is called when the record sent to the server has been acknowledged, or when sending the record fails before
@@ -86,10 +86,10 @@ public interface ProducerInterceptor<K, V> extends Configurable {
      *                 {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
      * @param exception The exception thrown during processing of this record. Null if no error occurred.
      */
-    public void onAcknowledgement(RecordMetadata metadata, Exception exception);
+    void onAcknowledgement(RecordMetadata metadata, Exception exception);
 
     /**
      * This is called when interceptor is closed
      */
-    public void close();
+    void close();
 }

clients/src/main/java/org/apache/kafka/common/compress/ZstdFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.kafka.common.compress;
 
+import com.github.luben.zstd.BufferPool;
 import com.github.luben.zstd.RecyclingBufferPool;
 import com.github.luben.zstd.ZstdInputStreamNoFinalizer;
 import com.github.luben.zstd.ZstdOutputStreamNoFinalizer;
@@ -47,10 +48,24 @@ public static OutputStream wrapForOutput(ByteBufferOutputStream buffer) {
 
     public static InputStream wrapForInput(ByteBuffer buffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
         try {
+            // We use our own BufferSupplier instead of com.github.luben.zstd.RecyclingBufferPool since our
+            // implementation doesn't require locking or soft references.
+            BufferPool bufferPool = new BufferPool() {
+                @Override
+                public ByteBuffer get(int capacity) {
+                    return decompressionBufferSupplier.get(capacity);
+                }
+
+                @Override
+                public void release(ByteBuffer buffer) {
+                    decompressionBufferSupplier.release(buffer);
+                }
+            };
+
             // Set output buffer (uncompressed) to 16 KB (none by default) to ensure reasonable performance
             // in cases where the caller reads a small number of bytes (potentially a single byte).
             return new BufferedInputStream(new ZstdInputStreamNoFinalizer(new ByteBufferInputStream(buffer),
-                RecyclingBufferPool.INSTANCE), 16 * 1024);
+                bufferPool), 16 * 1024);
         } catch (Throwable e) {
             throw new KafkaException(e);
         }

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

@@ -32,6 +32,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * A convenient base class for configurations to extend.
@@ -42,8 +43,12 @@ public class AbstractConfig {
 
     private final Logger log = LoggerFactory.getLogger(getClass());
 
-    /* configs for which values have been requested, used to detect unused configs */
-    private final Set<String> used;
+    /**
+     * Configs for which values have been requested, used to detect unused configs.
+     * This set must be concurrent modifiable and iterable. It will be modified
+     * when directly accessed or as a result of RecordingMap access.
+     */
+    private final Set<String> used = ConcurrentHashMap.newKeySet();
 
     /* the original values passed in by the user */
     private final Map<String, ?> originals;
@@ -106,7 +111,6 @@ public AbstractConfig(ConfigDef definition, Map<?, ?> originals,  Map<String, ?>
 
         this.originals = resolveConfigVariables(configProviderProps, (Map<String, Object>) originals);
         this.values = definition.parse(this.originals);
-        this.used = Collections.synchronizedSet(new HashSet<>());
         Map<String, Object> configUpdates = postProcessParsedConfig(Collections.unmodifiableMap(this.values));
         for (Map.Entry<String, Object> update : configUpdates.entrySet()) {
             this.values.put(update.getKey(), update.getValue());

clients/src/main/java/org/apache/kafka/server/authorizer/Action.java

@@ -31,27 +31,32 @@ public class Action {
     private final boolean logIfAllowed;
     private final boolean logIfDenied;
 
+    /**
+     * @param operation non-null operation being performed
+     * @param resourcePattern non-null resource pattern on which this action is being performed
+     */
     public Action(AclOperation operation,
                   ResourcePattern resourcePattern,
                   int resourceReferenceCount,
                   boolean logIfAllowed,
                   boolean logIfDenied) {
-        this.operation = operation;
-        this.resourcePattern = resourcePattern;
+        this.operation = Objects.requireNonNull(operation, "operation can't be null");
+        this.resourcePattern = Objects.requireNonNull(resourcePattern, "resourcePattern can't be null");
         this.logIfAllowed = logIfAllowed;
         this.logIfDenied = logIfDenied;
         this.resourceReferenceCount = resourceReferenceCount;
     }
 
     /**
-     * Resource on which action is being performed.
+     * @return a non-null resource pattern on which this action is being performed
      */
     public ResourcePattern resourcePattern() {
         return resourcePattern;
     }
 
     /**
-     * Operation being performed.
+     *
+     * @return a non-null operation being performed
      */
     public AclOperation operation() {
         return operation;

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -5773,7 +5773,7 @@ public void testClientSideTimeoutAfterFailureToReceiveResponse() throws Exceptio
             TestUtils.waitForCondition(() -> {
                 time.sleep(1);
                 return disconnectFuture.isDone();
-            }, 1, 5000, () -> "Timed out waiting for expected disconnect");
+            }, 5000, 1, () -> "Timed out waiting for expected disconnect");
             assertFalse(disconnectFuture.isCompletedExceptionally());
             assertFalse(result.future.isDone());
             TestUtils.waitForCondition(env.kafkaClient()::hasInFlightRequests,

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

@@ -1757,6 +1757,28 @@ private void consumerCloseTest(final long closeTimeoutMs,
         }
     }
 
+    @Test
+    public void testPartitionsForNonExistingTopic() {
+        Time time = new MockTime();
+        SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST);
+        ConsumerMetadata metadata = createMetadata(subscription);
+        MockClient client = new MockClient(time, metadata);
+
+        initMetadata(client, Collections.singletonMap(topic, 1));
+        Cluster cluster = metadata.fetch();
+
+        MetadataResponse updateResponse = RequestTestUtils.metadataResponse(cluster.nodes(),
+            cluster.clusterResource().clusterId(),
+            cluster.controller().id(),
+            Collections.emptyList());
+        client.prepareResponse(updateResponse);
+
+        ConsumerPartitionAssignor assignor = new RoundRobinAssignor();
+
+        KafkaConsumer<String, String> consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId);
+        assertEquals(Collections.emptyList(), consumer.partitionsFor("non-exist-topic"));
+    }
+
     @Test
     public void testPartitionsForAuthenticationFailure() {
         final KafkaConsumer<String, String> consumer = consumerWithPendingAuthenticationError();

clients/src/test/java/org/apache/kafka/test/TestUtils.java

@@ -298,7 +298,7 @@ public static void waitForCondition(final TestCondition testCondition, final lon
      * avoid transient failures due to slow or overloaded machines.
      */
     public static void waitForCondition(final TestCondition testCondition, final long maxWaitMs, Supplier<String> conditionDetailsSupplier) throws InterruptedException {
-        waitForCondition(testCondition, DEFAULT_POLL_INTERVAL_MS, maxWaitMs, conditionDetailsSupplier);
+        waitForCondition(testCondition, maxWaitMs, DEFAULT_POLL_INTERVAL_MS, conditionDetailsSupplier);
     }
 
     /**
@@ -310,11 +310,11 @@ public static void waitForCondition(final TestCondition testCondition, final lon
      */
     public static void waitForCondition(
         final TestCondition testCondition,
-        final long pollIntervalMs,
         final long maxWaitMs,
+        final long pollIntervalMs,
         Supplier<String> conditionDetailsSupplier
     ) throws InterruptedException {
-        retryOnExceptionWithTimeout(pollIntervalMs, maxWaitMs, () -> {
+        retryOnExceptionWithTimeout(maxWaitMs, pollIntervalMs, () -> {
             String conditionDetailsSupplied = conditionDetailsSupplier != null ? conditionDetailsSupplier.get() : null;
             String conditionDetails = conditionDetailsSupplied != null ? conditionDetailsSupplied : "";
             assertTrue(testCondition.conditionMet(),
@@ -333,7 +333,7 @@ public static void waitForCondition(
      */
     public static void retryOnExceptionWithTimeout(final long timeoutMs,
                                                    final ValuelessCallable runnable) throws InterruptedException {
-        retryOnExceptionWithTimeout(DEFAULT_POLL_INTERVAL_MS, timeoutMs, runnable);
+        retryOnExceptionWithTimeout(timeoutMs, DEFAULT_POLL_INTERVAL_MS, runnable);
     }
 
     /**
@@ -345,21 +345,21 @@ public static void retryOnExceptionWithTimeout(final long timeoutMs,
      * @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
      */
     public static void retryOnExceptionWithTimeout(final ValuelessCallable runnable) throws InterruptedException {
-        retryOnExceptionWithTimeout(DEFAULT_POLL_INTERVAL_MS, DEFAULT_MAX_WAIT_MS, runnable);
+        retryOnExceptionWithTimeout(DEFAULT_MAX_WAIT_MS, DEFAULT_POLL_INTERVAL_MS, runnable);
     }
 
     /**
      * Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
      * {@link AssertionError}s, or for the given timeout to expire. If the timeout expires then the
      * last exception or assertion failure will be thrown thus providing context for the failure.
      *
-     * @param pollIntervalMs the interval in milliseconds to wait between invoking {@code runnable}.
      * @param timeoutMs the total time in milliseconds to wait for {@code runnable} to complete successfully.
+     * @param pollIntervalMs the interval in milliseconds to wait between invoking {@code runnable}.
      * @param runnable the code to attempt to execute successfully.
      * @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
      */
-    public static void retryOnExceptionWithTimeout(final long pollIntervalMs,
-                                                   final long timeoutMs,
+    public static void retryOnExceptionWithTimeout(final long timeoutMs,
+                                                   final long pollIntervalMs,
                                                    final ValuelessCallable runnable) throws InterruptedException {
         final long expectedEnd = System.currentTimeMillis() + timeoutMs;
 

connect/mirror/src/test/java/org/apache/kafka/connect/mirror/CheckpointTest.java

@@ -32,9 +32,13 @@ public void testSerde() {
         byte[] value = checkpoint.recordValue();
         ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>("any-topic", 7, 8, key, value);
         Checkpoint deserialized = Checkpoint.deserializeRecord(record);
-        assertEquals(checkpoint.consumerGroupId(), deserialized.consumerGroupId());
-        assertEquals(checkpoint.topicPartition(), deserialized.topicPartition());
-        assertEquals(checkpoint.upstreamOffset(), deserialized.upstreamOffset());
-        assertEquals(checkpoint.downstreamOffset(), deserialized.downstreamOffset());
+        assertEquals(checkpoint.consumerGroupId(), deserialized.consumerGroupId(),
+                "Failure on checkpoint consumerGroupId serde");
+        assertEquals(checkpoint.topicPartition(), deserialized.topicPartition(),
+                "Failure on checkpoint topicPartition serde");
+        assertEquals(checkpoint.upstreamOffset(), deserialized.upstreamOffset(),
+                "Failure on checkpoint upstreamOffset serde");
+        assertEquals(checkpoint.downstreamOffset(), deserialized.downstreamOffset(),
+                "Failure on checkpoint downstreamOffset serde");
     }
 }