lianetm commented on code in PR #18089:
URL: https://github.com/apache/kafka/pull/18089#discussion_r1880554051


##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1175,21 +1175,8 @@ private CompletableFuture<Void> assignPartitions(
 
         // Invoke user call back.
         CompletableFuture<Void> result = 
signalPartitionsAssigned(addedPartitions);
-        result.whenComplete((__, exception) -> {
-            if (exception == null) {
-                // Enable newly added partitions to start fetching and 
updating positions for them.
-                
subscriptions.enablePartitionsAwaitingCallback(addedPartitions);
-            } else {
-                // Keeping newly added partitions as non-fetchable after the 
callback failure.
-                // They will be retried on the next reconciliation loop, until 
it succeeds or the
-                // broker removes them from the assignment.
-                if (!addedPartitions.isEmpty()) {
-                    log.warn("Leaving newly assigned partitions {} marked as 
non-fetchable and not " +
-                            "requiring initializing positions after 
onPartitionsAssigned callback failed.",
-                        addedPartitions, exception);
-                }
-            }
-        });
+        // Enable newly added partitions to start fetching and updating 
positions for them.
+        result.whenComplete((__, exception) -> 
subscriptions.enablePartitionsAwaitingCallback(addedPartitions));

Review Comment:
   my concern with the change that we currently have in this PR is that it 
seems to me it's allowing to fetch records even if callbacks always fail, 
doesn't it? (vs. allow fetch if the callback recovers)
   
   I expect the classic does not allow fetch if the callback always fail, 
because it will always break the poll loop when polling the coordinator, before 
making it to the fetch part. 
   
   The classic consumer allows fetch after a failed callback only if the 
callback recovers (and this is what the new consumer doesn't achieve because in 
that case t only re-enables added instead of assigned).
   
   Sorry if I'm missing something, thoughts? 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to