KAFKA-8421: Still return data during rebalance

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

@@ -1333,13 +1333,12 @@ private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(Timer tim
      *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
      *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
      *             consumer is for example no longer part of the group this exception would be thrown.
-     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the commit failed but can be retried.
-     *            This can occur if, e.g. consumer instance is in the middle of a rebalance so it is not yet determined
-     *            which partitions would be assigned to the consumer yet. In such cases you can first complete the rebalance
-     *            by calling {@link #poll(Duration)} and retry committing offsets again. NOTE when you retry after the
-     *            rebalance the assigned partitions may have changed, and also for those partitions that are still assigned
-     *            their fetch positions may have changed too if more records are returned from the {@link #poll(Duration)}
-     *            call, so when you retry committing only those assigned partitions would be committed with the current positions.
+     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
+     *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
+     *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
+     *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
+     *            and also for those partitions that are still assigned their fetch positions may have changed too
+     *            if more records are returned from the {@link #poll(Duration)} call.
      * @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
@@ -1377,13 +1376,12 @@ public void commitSync() {
      *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
      *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
      *             consumer is for example no longer part of the group this exception would be thrown.
-     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the commit failed because
-     *            it is in the middle of a rebalance so it is not yet determined
-     *            which partitions would be assigned to the consumer yet. In such cases you can first complete the rebalance
-     *            by calling {@link #poll(Duration)} and retry committing offsets again. NOTE when you retry after the
-     *            rebalance the assigned partitions may have changed, and also for those partitions that are still assigned
-     *            their fetch positions may have changed too if more records are returned from the {@link #poll(Duration)}
-     *            call, so when you retry committing only those assigned partitions would be committed with the current positions.
+     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
+     *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
+     *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
+     *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
+     *            and also for those partitions that are still assigned their fetch positions may have changed too
+     *            if more records are returned from the {@link #poll(Duration)} call.
      * @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
@@ -1433,14 +1431,13 @@ public void commitSync(Duration timeout) {
      *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
      *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
      *             consumer is for example no longer part of the group this exception would be thrown.
-     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the commit failed because
-     *            it is in the middle of a rebalance so it is not yet determined
-     *            which partitions would be assigned to the consumer yet. In such cases you can first complete the rebalance
-     *            by calling {@link #poll(Duration)} and retry committing offsets again. NOTE when you retry after the
-     *            rebalance the assigned partitions may have changed, and also for those partitions that are still assigned
-     *            their fetch positions may also have changed too if more records are returned from the {@link #poll(Duration)}
-     *            call, so when you retry committing you should include only those assigned partitions after the rebalance
-     *            with the current fetch position; otherwise a fatal {@link CommitFailedException} will be thrown.
+     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
+     *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
+     *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
+     *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
+     *            and also for those partitions that are still assigned their fetch positions may have changed too
+     *            if more records are returned from the {@link #poll(Duration)} call, so when you retry committing
+     *            you should consider updating the passed in {@code offset} parameter.
      * @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
@@ -1482,14 +1479,13 @@ public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets) {
      *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
      *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
      *             consumer is for example no longer part of the group this exception would be thrown.
-     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the commit failed because
-     *            it is in the middle of a rebalance so it is not yet determined
-     *            which partitions would be assigned to the consumer yet. In such cases you can first complete the rebalance
-     *            by calling {@link #poll(Duration)} and retry committing offsets again. NOTE when you retry after the
-     *            rebalance the assigned partitions may have changed, and also for those partitions that are still assigned
-     *            their fetch positions may also have changed too if more records are returned from the {@link #poll(Duration)}
-     *            call, so when you retry committing you should include only those assigned partitions after the rebalance
-     *            with the current fetch position; otherwise a fatal {@link CommitFailedException} will be thrown.
+     * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
+     *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
+     *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
+     *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
+     *            and also for those partitions that are still assigned their fetch positions may have changed too
+     *            if more records are returned from the {@link #poll(Duration)} call, so when you retry committing
+     *            you should consider updating the passed in {@code offset} parameter.
      * @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
@@ -1582,9 +1578,6 @@ public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, Of
             log.debug("Committing offsets: {}", offsets);
             offsets.forEach(this::updateLastSeenEpochIfNewer);
             coordinator.commitOffsetsAsync(new HashMap<>(offsets), callback);
-        } catch (CommitFailedException e) {
-            log.error("Failed to commit offsets asynchronously because they do not belong to dynamically assigned partitions");
-            callback.onComplete(offsets, e);
         } finally {
             release();
         }

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

@@ -22,8 +22,22 @@ public class RetriableCommitFailedException extends RetriableException {
 
     private static final long serialVersionUID = 1L;
 
+    public static RetriableCommitFailedException withUnderlyingMessage(String additionalMessage) {
+        return new RetriableCommitFailedException("Offset commit failed with a retriable exception. " +
+                "You should retry committing the latest consumed offsets. " +
+                "The underlying error was: " + additionalMessage);
+    }
+
     public RetriableCommitFailedException(Throwable t) {
         super("Offset commit failed with a retriable exception. You should retry committing " +
                 "the latest consumed offsets.", t);
     }
+
+    public RetriableCommitFailedException(String message) {
+        super(message);
+    }
+
+    public RetriableCommitFailedException(String message, Throwable t) {
+        super(message, t);
+    }
 }

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

@@ -131,7 +131,7 @@ protected enum MemberState {
     private HeartbeatThread heartbeatThread = null;
     private RequestFuture<ByteBuffer> joinFuture = null;
     private RequestFuture<Void> findCoordinatorFuture = null;
-    private RuntimeException findCoordinatorException = null;
+    volatile private RuntimeException findCoordinatorException = null;
     private Generation generation = Generation.NO_GENERATION;
     private long lastRebalanceStartMs = -1L;
     private long lastRebalanceEndMs = -1L;
@@ -237,16 +237,6 @@ protected synchronized boolean ensureCoordinatorReady(final Timer timer) {
 
             if (!future.isDone()) {
                 // ran out of time
-                future.addListener(new RequestFutureListener<Void>() {
-                    @Override
-                    public void onSuccess(Void value) {} // do nothing
-
-                    @Override
-                    public void onFailure(RuntimeException e) {
-                        findCoordinatorException = e;
-                    }
-                });
-
                 break;
             }
 
@@ -274,8 +264,20 @@ protected synchronized RequestFuture<Void> lookupCoordinator() {
             if (node == null) {
                 log.debug("No broker available to send FindCoordinator request");
                 return RequestFuture.noBrokersAvailable();
-            } else
+            } else{
                 findCoordinatorFuture = sendFindCoordinatorRequest(node);
+                // remember the exception even after the future is cleared so that
+                // it can still be thrown by the ensureCoordinatorReady caller
+                findCoordinatorFuture.addListener(new RequestFutureListener<Void>() {
+                    @Override
+                    public void onSuccess(Void value) {} // do nothing
+
+                    @Override
+                    public void onFailure(RuntimeException e) {
+                        findCoordinatorException = e;
+                    }
+                });
+            }
         }
         return findCoordinatorFuture;
     }

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

@@ -769,6 +769,9 @@ public boolean refreshCommittedOffsetsIfNeeded(Timer timer) {
                     this.subscriptions.seekUnvalidated(tp, position);
 
                     log.info("Setting offset for partition {} to the committed offset {}", tp, position);
+                } else {
+                    log.info("Ignoring the returned {} since its partition {} is no longer assigned",
+                        offsetAndMetadata, tp);
                 }
             }
         }

clients/src/main/java/org/apache/kafka/common/errors/RebalanceInProgressException.java

@@ -19,7 +19,19 @@
 public class RebalanceInProgressException extends ApiException {
     private static final long serialVersionUID = 1L;
 
+    public RebalanceInProgressException() {
+        super();
+    }
+
+    public RebalanceInProgressException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
     public RebalanceInProgressException(String message) {
         super(message);
     }
+
+    public RebalanceInProgressException(Throwable cause) {
+        super(cause);
+    }
 }