KAFKA-20684 [7/N]: Point clients internals at RebalanceListener

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

@@ -151,9 +151,9 @@ public synchronized void rebalance(Collection<TopicPartition> newAssignment) {
     }
 
     /**
-     * Simulates a partition loss event. Calls {@link ConsumerRebalanceListener#onPartitionsLost}
+     * Simulates a partition loss event. Calls {@link RebalanceListener#onPartitionsLost}
      * for the specified partitions and removes them from the current assignment. Unlike
-     * {@link #rebalance(Collection)}, which calls {@link ConsumerRebalanceListener#onPartitionsRevoked},
+     * {@link #rebalance(Collection)}, which calls {@link RebalanceListener#onPartitionsRevoked},
      * this method models the case where the consumer loses partitions without a graceful revoke..
      *
      * <p>Only records belonging to the lost partitions are cleared; records for retained
@@ -222,7 +222,7 @@ public void subscribe(Collection<String> topics, final ConsumerRebalanceListener
         subscribeInternal(topics, listener);
     }
 
-    private synchronized void subscribeInternal(SubscriptionPattern pattern, ConsumerRebalanceListener listener) {
+    private synchronized void subscribeInternal(SubscriptionPattern pattern, RebalanceListener listener) {
         if (pattern == null || pattern.toString().isEmpty())
             throw new IllegalArgumentException("Topic pattern cannot be " + (pattern == null ? "null" : "empty"));
 
@@ -233,15 +233,15 @@ private synchronized void subscribeInternal(SubscriptionPattern pattern, Consume
         subscriptions.subscribe(pattern);
     }
 
-    private synchronized void subscribeInternal(Collection<String> topics, ConsumerRebalanceListener listener) {
+    private synchronized void subscribeInternal(Collection<String> topics, RebalanceListener listener) {
         ensureNotClosed();
         committed.clear();
         if (listener != null)
             subscriptions.setRebalanceListener(listener, this);
         subscriptions.subscribe(new HashSet<>(topics));
     }
 
-    private synchronized void subscribeInternal(Pattern pattern, ConsumerRebalanceListener listener) {
+    private synchronized void subscribeInternal(Pattern pattern, RebalanceListener listener) {
         ensureNotClosed();
         committed.clear();
         if (listener != null)

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

@@ -30,7 +30,7 @@
 import java.util.Set;
 
 /**
- * A restricted view of a {@link Consumer} passed to {@link ConsumerRebalanceListener} callback
+ * A restricted view of a {@link Consumer} passed to {@link RebalanceListener} callback
  * methods during a partition rebalance. This interface provides compile-time enforcement of safe
  * consumer operations during rebalance callbacks, replacing the previous pattern of capturing a
  * {@code Consumer} reference externally (e.g. via constructor injection), which gave callbacks
@@ -78,7 +78,7 @@
  *   <li>{@code enforceRebalance()} - would trigger re-entrant rebalance</li>
  * </ul>
  *
- * @see ConsumerRebalanceListener
+ * @see RebalanceListener
  */
 @InterfaceAudience.Public
 public interface RebalanceConsumer {

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

@@ -111,23 +111,23 @@
  * <li><code>C2 [t2p0, t2p1, t2p2]</code></li>
  * </ul>
  *</p>
- * <h3>Impact on <code>ConsumerRebalanceListener</code></h3>
+ * <h3>Impact on <code>RebalanceListener</code></h3>
  * The sticky assignment strategy can provide some optimization to those consumers that have some partition cleanup code
  * in their <code>onPartitionsRevoked()</code> callback listeners. The cleanup code is placed in that callback listener
  * because the consumer has no assumption or hope of preserving any of its assigned partitions after a rebalance when it
  * is using range or round robin assignor. The listener code would look like this:
  * <pre>
  * {@code
- * class TheOldRebalanceListener implements ConsumerRebalanceListener {
+ * class TheOldRebalanceListener implements RebalanceListener {
  *
- *   void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+ *   void onPartitionsRevoked(Collection<TopicPartition> partitions, RebalanceConsumer consumer) {
  *     for (TopicPartition partition: partitions) {
  *       commitOffsets(partition);
  *       cleanupState(partition);
  *     }
  *   }
  *
- *   void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+ *   void onPartitionsAssigned(Collection<TopicPartition> partitions, RebalanceConsumer consumer) {
  *     for (TopicPartition partition: partitions) {
  *       initializeState(partition);
  *       initializeOffset(partition);
@@ -145,15 +145,15 @@
  * clarifies this point:
  * <pre>
  * {@code
- * class TheNewRebalanceListener implements ConsumerRebalanceListener {
+ * class TheNewRebalanceListener implements RebalanceListener {
  *   Collection<TopicPartition> lastAssignment = Collections.emptyList();
  *
- *   void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+ *   void onPartitionsRevoked(Collection<TopicPartition> partitions, RebalanceConsumer consumer) {
  *     for (TopicPartition partition: partitions)
  *       commitOffsets(partition);
  *   }
  *
- *   void onPartitionsAssigned(Collection<TopicPartition> assignment) {
+ *   void onPartitionsAssigned(Collection<TopicPartition> assignment, RebalanceConsumer consumer) {
  *     for (TopicPartition partition: difference(lastAssignment, assignment))
  *       cleanupState(partition);
  *
@@ -170,7 +170,7 @@
  * </pre>
  *
  * Any consumer that uses sticky assignment can leverage this listener like this:
- * <code>consumer.subscribe(topics, new TheNewRebalanceListener());</code>
+ * <code>consumer.setRebalanceListener(new TheNewRebalanceListener());</code>
  *
  * Note that you can leverage the {@link CooperativeStickyAssignor} so that only partitions which are being
  * reassigned to another consumer will be revoked. That is the preferred assignor for newer cluster. See

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

@@ -19,7 +19,7 @@
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.consumer.CloseOptions;
 import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.RebalanceListener;
 import org.apache.kafka.clients.consumer.internals.metrics.RebalanceMetricsManager;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
@@ -108,7 +108,7 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
     protected final SubscriptionState subscriptions;
 
     /**
-     * Metadata that allows us to create the partitions needed for {@link ConsumerRebalanceListener}.
+     * Metadata that allows us to create the partitions needed for {@link RebalanceListener}.
      */
     private final Metadata metadata;
 
@@ -618,7 +618,7 @@ public CompletableFuture<Void> leaveGroup() {
      * transition to {@link MemberState#LEAVING} to send the heartbeat request and leave the group.
      * This is expected to be invoked when the user calls the unsubscribe API or is closing the consumer.
      *
-     * @param runCallbacks {@code true} to insert the step to execute the {@link ConsumerRebalanceListener} callback,
+     * @param runCallbacks {@code true} to insert the step to execute the {@link RebalanceListener} callback,
      *                     {@code false} to skip
      *
      * @return Future that will complete when the callback execution completes and the heartbeat

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

@@ -186,7 +186,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
      *
      * <ul>
      *     <li>Errors that occur in the network thread that need to be propagated to the application thread</li>
-     *     <li>{@link ConsumerRebalanceListener} callbacks that are to be executed on the application thread</li>
+     *     <li>{@link RebalanceListener} callbacks that are to be executed on the application thread</li>
      * </ul>
      */
     private class BackgroundEventProcessor implements EventProcessor<BackgroundEvent> {
@@ -1582,37 +1582,37 @@ public void close(CloseOptions option) {
      *
      * <ol>
      *     <li>
-     *         The execution of the {@link ConsumerRebalanceListener} callback (if applicable) must be performed on
+     *         The execution of the {@link RebalanceListener} callback (if applicable) must be performed on
      *         the application thread to ensure it does not interfere with the network I/O on the background thread.
      *     </li>
      *     <li>
-     *         The {@link ConsumerRebalanceListener} callback execution must complete before an attempt to leave
+     *         The {@link RebalanceListener} callback execution must complete before an attempt to leave
      *         the consumer group is performed. In this context, “complete” does not necessarily imply
      *         <em>success</em>; execution is “complete” even if the execution <em>fails</em> with an error.
      *     </li>
      *     <li>
-     *         Any error thrown during the {@link ConsumerRebalanceListener} callback execution will be caught to
+     *         Any error thrown during the {@link RebalanceListener} callback execution will be caught to
      *         ensure it does not prevent execution of the remaining {@link #close()} logic.
      *     </li>
      *     <li>
      *         The application thread will be blocked during the entire duration of the execution of the
-     *         {@link ConsumerRebalanceListener}. The consumer does not employ a mechanism to short-circuit the
+     *         {@link RebalanceListener}. The consumer does not employ a mechanism to short-circuit the
      *         callback execution, so execution is not bound by the timeout in {@link #close(Duration)}.
      *     </li>
      *     <li>
-     *         A given {@link ConsumerRebalanceListener} implementation may be affected by the application thread's
+     *         A given {@link RebalanceListener} implementation may be affected by the application thread's
      *         interrupt state. If the callback implementation performs any blocking operations, it may result in
      *         an error. An implementation may choose to preemptively check the thread's interrupt flag via
      *         {@link Thread#isInterrupted()} or {@link Thread#isInterrupted()} and alter its behavior.
      *     </li>
      *     <li>
      *         If the application thread was interrupted <em>prior</em> to the execution of the
-     *         {@link ConsumerRebalanceListener} callback, the thread's interrupt state will be preserved for the
-     *         {@link ConsumerRebalanceListener} execution.
+     *         {@link RebalanceListener} callback, the thread's interrupt state will be preserved for the
+     *         {@link RebalanceListener} execution.
      *     </li>
      *     <li>
      *         If the application thread was interrupted <em>prior</em> to the execution of the
-     *         {@link ConsumerRebalanceListener} callback <em>but</em> the callback cleared out the interrupt state,
+     *         {@link RebalanceListener} callback <em>but</em> the callback cleared out the interrupt state,
      *         the {@link #close()} method will not make any effort to restore the application thread's interrupt
      *         state for the remainder of the execution of {@link #close()}.
      *     </li>
@@ -2253,7 +2253,7 @@ private void release() {
             currentThread.set(NO_CURRENT_THREAD);
     }
 
-    private void subscribeInternal(Pattern pattern, ConsumerRebalanceListener listener) {
+    private void subscribeInternal(Pattern pattern, RebalanceListener listener) {
         acquireAndEnsureOpen();
         subscriptions.setRebalanceListener(listener, this);
         try {
@@ -2276,7 +2276,7 @@ private void subscribeInternal(Pattern pattern, ConsumerRebalanceListener listen
      * subscription state, so it's included in the next heartbeat request sent to the broker.
      * No validation of the pattern is performed by the client (other than null/empty checks).
      */
-    private void subscribeToRegex(SubscriptionPattern pattern, ConsumerRebalanceListener listener) {
+    private void subscribeToRegex(SubscriptionPattern pattern, RebalanceListener listener) {
         acquireAndEnsureOpen();
         try {
             throwIfGroupIdNotDefined();
@@ -2301,7 +2301,7 @@ private void throwIfSubscriptionPatternIsInvalid(SubscriptionPattern subscriptio
         }
     }
 
-    private void subscribeInternal(Collection<String> topics, ConsumerRebalanceListener listener) {
+    private void subscribeInternal(Collection<String> topics, RebalanceListener listener) {
         acquireAndEnsureOpen();
         try {
             throwIfGroupIdNotDefined();
@@ -2436,7 +2436,7 @@ boolean processBackgroundEvents(boolean skipAssignmentEvents) {
      * As an example, take {@link #unsubscribe()}. To start unsubscribing, the application thread enqueues an
      * {@link UnsubscribeEvent} on the application event queue. That event will eventually trigger the
      * rebalancing logic in the background thread. Critically, as part of this rebalancing work, the
-     * {@link ConsumerRebalanceListener#onPartitionsRevoked(Collection)} callback needs to be invoked for any
+     * {@link RebalanceListener#onPartitionsRevoked(Collection, RebalanceConsumer)} callback needs to be invoked for any
      * partitions the consumer owns. However,
      * this callback must be executed on the application thread. To achieve this, the background thread enqueues a
      * {@link PartitionsRemovedEvent} on its background event queue. That event queue is
@@ -2445,7 +2445,7 @@ boolean processBackgroundEvents(boolean skipAssignmentEvents) {
      * {@link ConsumerRebalanceListenerCallbackCompletedEvent} is then enqueued by the application thread on the
      * application event queue. Moments later, the background thread will see that event, process it, and continue
      * execution of the rebalancing logic. The rebalancing logic cannot complete until the
-     * {@link ConsumerRebalanceListener} callback is performed.
+     * {@link RebalanceListener} callback is performed.
      *
      * @param future                    Event that contains a {@link CompletableFuture}; it is on this future that the
      *                                  application thread will wait for completion

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java

@@ -488,7 +488,7 @@ public void subscribe(Collection<String> topics) {
      *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
      *                               configured at-least one partition assignment strategy
      */
-    private void subscribeInternal(Collection<String> topics, ConsumerRebalanceListener listener) {
+    private void subscribeInternal(Collection<String> topics, RebalanceListener listener) {
         acquireAndEnsureOpen();
         try {
             throwIfGroupIdNotDefined();
@@ -562,7 +562,7 @@ public void subscribe(SubscriptionPattern pattern) {
      * the max metadata age, the consumer will refresh metadata more often and check for matching topics.
      * <p>
      * See {@link #subscribe(Collection, ConsumerRebalanceListener)} for details on the
-     * use of the {@link ConsumerRebalanceListener}. Generally rebalances are triggered when there
+     * use of the {@link RebalanceListener}. Generally rebalances are triggered when there
      * is a change to the topics matching the provided pattern and when consumer group membership changes.
      * Group rebalances only take place during an active call to {@link #poll(Duration)}.
      *
@@ -574,7 +574,7 @@ public void subscribe(SubscriptionPattern pattern) {
      *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
      *                               configured at-least one partition assignment strategy
      */
-    private void subscribeInternal(Pattern pattern, ConsumerRebalanceListener listener) {
+    private void subscribeInternal(Pattern pattern, RebalanceListener listener) {
         throwIfGroupIdNotDefined();
         if (pattern == null || pattern.toString().isEmpty())
             throw new IllegalArgumentException("Topic pattern to subscribe to cannot be " + (pattern == null ?

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

@@ -154,7 +154,7 @@ private boolean sameRequest(final Set<TopicPartition> currentRequest, final Gene
     }
 
     private final RebalanceProtocol protocol;
-    // Wraps the logic for invoking the ConsumerRebalanceListener methods
+    // Wraps the logic for invoking the RebalanceListener methods
     private final ConsumerRebalanceListenerInvoker rebalanceListenerInvoker;
     // pending commit offset request in onJoinPrepare
     private RequestFuture<Void> autoCommitOffsetRequestFuture = null;

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java

@@ -18,7 +18,7 @@
 
 import org.apache.kafka.clients.consumer.CloseOptions;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.RebalanceListener;
 import org.apache.kafka.clients.consumer.internals.events.ApplyAssignmentEvent;
 import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
 import org.apache.kafka.clients.consumer.internals.events.CompletableBackgroundEvent;
@@ -400,7 +400,7 @@ public boolean isLeavingGroup() {
 
     /**
      * Enqueue a {@link PartitionsRemovedEvent} to trigger the execution of either
-     * {@link ConsumerRebalanceListener#onPartitionsRevoked} or {@link ConsumerRebalanceListener#onPartitionsLost}
+     * {@link RebalanceListener#onPartitionsRevoked} or {@link RebalanceListener#onPartitionsLost}
      * on the application thread.
      *
      * <p/>
@@ -441,9 +441,9 @@ private CompletableFuture<Void> enqueuePartitionsAssignedEvent(Set<TopicPartitio
     }
 
     /**
-     * Signals that a {@link ConsumerRebalanceListener} callback has completed. This is invoked when the
+     * Signals that a {@link RebalanceListener} callback has completed. This is invoked when the
      * application thread has completed the callback and has submitted a
-     * {@link ConsumerRebalanceListenerCallbackCompletedEvent} to the network I/O thread. At this point, we
+     * {@link RebalanceListenerCallbackCompletedEvent} to the network I/O thread. At this point, we
      * notify the state machine that it's complete so that it can move to the next appropriate step of the
      * rebalance process.
      *

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java

@@ -30,7 +30,7 @@
 
 /**
  * This class encapsulates the invocation of the callback methods defined in the
- * {@link org.apache.kafka.clients.consumer.ConsumerRebalanceListener} interface. When consumer
+ * {@link org.apache.kafka.clients.consumer.RebalanceListener} interface. When consumer
  * group partition assignment changes, these methods are invoked. This class wraps those callback
  * calls with logging, metrics recording, and exception handling.
  */

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerMethodName.java

@@ -16,10 +16,10 @@
  */
 package org.apache.kafka.clients.consumer.internals;
 
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.RebalanceListener;
 
 /**
- * This class just provides a static name for the methods in the {@link ConsumerRebalanceListener} interface
+ * This class just provides a static name for the methods in the {@link RebalanceListener} interface
  * for a bit more compile time assurance.
  */
 public enum ConsumerRebalanceListenerMethodName {
@@ -31,11 +31,11 @@ public enum ConsumerRebalanceListenerMethodName {
     private final String fullyQualifiedMethodName;
 
     ConsumerRebalanceListenerMethodName(String methodName) {
-        this.fullyQualifiedMethodName = String.format("%s.%s", ConsumerRebalanceListener.class.getSimpleName(), methodName);
+        this.fullyQualifiedMethodName = String.format("%s.%s", RebalanceListener.class.getSimpleName(), methodName);
     }
 
     /**
-     * Provides the fully-qualified method name, e.g. {@code ConsumerRebalanceListener.onPartitionsRevoked}. This
+     * Provides the fully-qualified method name, e.g. {@code RebalanceListener.onPartitionsRevoked}. This
      * is used for log messages.
      *
      * @return Fully-qualified method name