You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2024/02/12 14:43:29 UTC
(kafka) branch trunk updated: KAFKA-15832: Trigger client reconciliation based on manager poll (#15275)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new c6f4c604d8e KAFKA-15832: Trigger client reconciliation based on manager poll (#15275)
c6f4c604d8e is described below
commit c6f4c604d8e50ad9e182eeb66f0d1650aa44f277
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Mon Feb 12 15:43:21 2024 +0100
KAFKA-15832: Trigger client reconciliation based on manager poll (#15275)
Currently, the reconciliation logic on the client is triggered when a new target assignment is received and resolved, or when new unresolved target assignments are discovered in metadata.
This change improves this by triggering the reconciliation logic on each poll iteration, to reconcile whatever is ready to be reconciled. This would require changes to support poll on the MembershipManager, and integrate it with the current polling logic in the background thread. Receiving a new target assignment from the broker, or resolving new topic names via a metadata update could only ensure that the #assignmentReadyToReconcile is properly updated (currently done), but wouldn't [...]
Reviewers: David Jacot <dj...@confluent.io>, Lianet Magrans <li...@gmail.com>
---
.../consumer/internals/MembershipManager.java | 2 +-
.../consumer/internals/MembershipManagerImpl.java | 76 +++-----
.../consumer/internals/RequestManagers.java | 9 +-
.../consumer/internals/ConsumerTestBuilder.java | 4 +-
.../internals/HeartbeatRequestManagerTest.java | 4 +-
.../internals/MembershipManagerImplTest.java | 200 +++++++++++++++++++--
.../events/ApplicationEventProcessorTest.java | 4 +-
7 files changed, 225 insertions(+), 74 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
index f2e1291a52e..7f0975b5aa9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
@@ -34,7 +34,7 @@ import java.util.concurrent.CompletableFuture;
* <li>Keeping assignment for the member</li>
* <li>Computing assignment for the group if the member is required to do so<li/>
*/
-public interface MembershipManager {
+public interface MembershipManager extends RequestManager {
/**
* @return Group ID of the consumer group the member is part of (or wants to be part of).
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
index 50f9a920a56..a5a28da049e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
@@ -19,6 +19,7 @@ package org.apache.kafka.clients.consumer.internals;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult;
import org.apache.kafka.clients.consumer.internals.Utils.TopicIdPartitionComparator;
import org.apache.kafka.clients.consumer.internals.Utils.TopicPartitionComparator;
import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
@@ -26,8 +27,6 @@ import org.apache.kafka.clients.consumer.internals.events.CompletableBackgroundE
import org.apache.kafka.clients.consumer.internals.events.ConsumerRebalanceListenerCallbackCompletedEvent;
import org.apache.kafka.clients.consumer.internals.events.ConsumerRebalanceListenerCallbackNeededEvent;
import org.apache.kafka.clients.consumer.internals.events.ErrorBackgroundEvent;
-import org.apache.kafka.common.ClusterResource;
-import org.apache.kafka.common.ClusterResourceListener;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
@@ -112,7 +111,7 @@ import static org.apache.kafka.clients.consumer.internals.ConsumerRebalanceListe
* the user if the callbacks fail. This manager is only concerned about the callbacks completion to
* know that it can proceed with the reconciliation.
*/
-public class MembershipManagerImpl implements MembershipManager, ClusterResourceListener {
+public class MembershipManagerImpl implements MembershipManager {
/**
* TopicPartition comparator based on topic name and partition id.
@@ -215,9 +214,9 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
*/
private final Map<Uuid, SortedSet<Integer>> currentTargetAssignment;
- /**
+ /**
* If there is a reconciliation running (triggering commit, callbacks) for the
- * assignmentReadyToReconcile. This will be true if {@link #reconcile()} has been triggered
+ * assignmentReadyToReconcile. This will be true if {@link #maybeReconcile()} has been triggered
* after receiving a heartbeat response, or a metadata update.
*/
private boolean reconciliationInProgress;
@@ -237,14 +236,6 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
*/
private Optional<CompletableFuture<Void>> leaveGroupInProgress = Optional.empty();
- /**
- * True if the member has registered to be notified when the cluster metadata is updated.
- * This is initially false, as the member that is not part of a consumer group does not
- * require metadata updated. This becomes true the first time the member joins on the
- * {@link #transitionToJoining()}
- */
- private boolean isRegisteredForMetadataUpdates;
-
/**
* Registered listeners that will be notified whenever the memberID/epoch gets updated (valid
* values received from the broker, or values cleared due to member leaving the group, getting
@@ -393,9 +384,9 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
/**
* This will process the assignment received if it is different from the member's current
- * assignment. If a new assignment is received, this will try to resolve the topic names from
- * metadata, reconcile the resolved assignment, and keep the unresolved to be reconciled when
- * metadata is discovered.
+ * assignment. If a new assignment is received, this will make sure reconciliation is attempted
+ * on the next call of `poll`. If another reconciliation is currently in process, the first `poll`
+ * after that reconciliation will trigger the new reconciliation.
*
* @param assignment Assignment received from the broker.
*/
@@ -406,7 +397,6 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
// assignment from the broker, different from the current assignment. Note that the
// reconciliation might not be triggered just yet because of missing metadata.
transitionTo(MemberState.RECONCILING);
- reconcile();
} else {
// Same assignment received, nothing to reconcile.
log.debug("Target assignment {} received from the broker is equals to the member " +
@@ -567,18 +557,6 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
resetEpoch();
transitionTo(MemberState.JOINING);
clearPendingAssignmentsAndLocalNamesCache();
- registerForMetadataUpdates();
- }
-
- /**
- * Register to get notified when the cluster metadata is updated, via the
- * {@link #onUpdate(ClusterResource)}. Register only if the manager is not register already.
- */
- private void registerForMetadataUpdates() {
- if (!isRegisteredForMetadataUpdates) {
- this.metadata.addClusterUpdateListener(this);
- isRegisteredForMetadataUpdates = true;
- }
}
/**
@@ -770,12 +748,15 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
* Reconcile the assignment that has been received from the server. If for some topics, the
* topic ID cannot be matched to a topic name, a metadata update will be triggered and only
* the subset of topics that are resolvable will be reconciled. Reconciliation will trigger the
- * callbacks and update the subscription state. Note that only one reconciliation
- * can be in progress at a time. If there is already another one in progress when this is
- * triggered, it will be no-op, and the assignment will be reconciled on the next
- * reconciliation loop.
+ * callbacks and update the subscription state.
+ *
+ * There are three conditions under which no reconciliation will be triggered:
+ * - We have already reconciled the assignment (the target assignment is the same as the current assignment).
+ * - Another reconciliation is already in progress.
+ * - There are topics that haven't been added to the current assignment yet, but all their topic IDs
+ * are missing from the target assignment.
*/
- void reconcile() {
+ void maybeReconcile() {
if (targetAssignmentReconciled()) {
log.debug("Ignoring reconciliation attempt. Target assignment is equal to the " +
"current assignment.");
@@ -1356,29 +1337,12 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
/**
* @return If there is a reconciliation in process now. Note that reconciliation is triggered
- * by a call to {@link #reconcile()}. Visible for testing.
+ * by a call to {@link #maybeReconcile()}. Visible for testing.
*/
boolean reconciliationInProgress() {
return reconciliationInProgress;
}
- /**
- * When cluster metadata is updated, try to resolve topic names for topic IDs received in
- * assignment that hasn't been resolved yet.
- * <ul>
- * <li>Try to find topic names for all assignments</li>
- * <li>Add discovered topic names to the local topic names cache</li>
- * <li>If any topics are resolved, trigger a reconciliation process</li>
- * <li>If some topics still remain unresolved, request another metadata update</li>
- * </ul>
- */
- @Override
- public void onUpdate(ClusterResource clusterResource) {
- if (state == MemberState.RECONCILING) {
- reconcile();
- }
- }
-
/**
* Register a new listener that will be invoked whenever the member state changes, or a new
* member ID or epoch is received.
@@ -1392,4 +1356,12 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
}
this.stateUpdatesListeners.add(listener);
}
+
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING) {
+ maybeReconcile();
+ }
+ return PollResult.EMPTY;
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
index b3c0c5b63b5..2d90a3ad708 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -48,6 +48,7 @@ public class RequestManagers implements Closeable {
public final Optional<CoordinatorRequestManager> coordinatorRequestManager;
public final Optional<CommitRequestManager> commitRequestManager;
public final Optional<HeartbeatRequestManager> heartbeatRequestManager;
+ public final Optional<MembershipManager> membershipManager;
public final OffsetsRequestManager offsetsRequestManager;
public final TopicMetadataRequestManager topicMetadataRequestManager;
public final FetchRequestManager fetchRequestManager;
@@ -60,7 +61,8 @@ public class RequestManagers implements Closeable {
FetchRequestManager fetchRequestManager,
Optional<CoordinatorRequestManager> coordinatorRequestManager,
Optional<CommitRequestManager> commitRequestManager,
- Optional<HeartbeatRequestManager> heartbeatRequestManager) {
+ Optional<HeartbeatRequestManager> heartbeatRequestManager,
+ Optional<MembershipManager> membershipManager) {
this.log = logContext.logger(RequestManagers.class);
this.offsetsRequestManager = requireNonNull(offsetsRequestManager, "OffsetsRequestManager cannot be null");
this.coordinatorRequestManager = coordinatorRequestManager;
@@ -68,11 +70,13 @@ public class RequestManagers implements Closeable {
this.topicMetadataRequestManager = topicMetadataRequestManager;
this.fetchRequestManager = fetchRequestManager;
this.heartbeatRequestManager = heartbeatRequestManager;
+ this.membershipManager = membershipManager;
List<Optional<? extends RequestManager>> list = new ArrayList<>();
list.add(coordinatorRequestManager);
list.add(commitRequestManager);
list.add(heartbeatRequestManager);
+ list.add(membershipManager);
list.add(Optional.of(offsetsRequestManager));
list.add(Optional.of(topicMetadataRequestManager));
list.add(Optional.of(fetchRequestManager));
@@ -205,7 +209,8 @@ public class RequestManagers implements Closeable {
fetch,
Optional.ofNullable(coordinator),
Optional.ofNullable(commit),
- Optional.ofNullable(heartbeatRequestManager)
+ Optional.ofNullable(heartbeatRequestManager),
+ Optional.ofNullable(membershipManager)
);
}
};
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java
index f6f71f3f7a9..d6ae6295060 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java
@@ -263,7 +263,9 @@ public class ConsumerTestBuilder implements Closeable {
fetchRequestManager,
coordinatorRequestManager,
commitRequestManager,
- heartbeatRequestManager);
+ heartbeatRequestManager,
+ membershipManager
+ );
this.applicationEventProcessor = spy(new ApplicationEventProcessor(
logContext,
applicationEventQueue,
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
index bc014efa773..4040d682331 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
@@ -546,7 +546,9 @@ public class HeartbeatRequestManagerTest {
.setAssignment(assignmentTopic1));
when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, "topic1"));
membershipManager.onHeartbeatResponseReceived(rs1.data());
- assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
+
+ // We remain in RECONCILING state, as the assignment will be reconciled on the next poll
+ assertEquals(MemberState.RECONCILING, membershipManager.state());
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
index 234b374d332..bb441fe7e64 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
@@ -57,6 +57,7 @@ import static org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer.inv
import static org.apache.kafka.common.requests.ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.apache.kafka.common.utils.Utils.mkSet;
import static org.apache.kafka.common.utils.Utils.mkSortedSet;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -165,7 +166,6 @@ public class MembershipManagerImplTest {
subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
backgroundEventHandler, time);
manager.transitionToJoining();
- verify(metadata).addClusterUpdateListener(manager);
clearInvocations(metadata);
// Following joins should not register again.
@@ -176,7 +176,6 @@ public class MembershipManagerImplTest {
manager.onHeartbeatRequestSent();
assertEquals(MemberState.UNSUBSCRIBED, manager.state());
manager.transitionToJoining();
- verify(metadata, never()).addClusterUpdateListener(manager);
}
@Test
@@ -507,12 +506,10 @@ public class MembershipManagerImplTest {
}
/**
- * This is the case where a member is stuck reconciling an assignment A (waiting on
- * metadata, commit or callbacks), and the target assignment changes (due to new topics added
- * to metadata, or new assignment received from broker). If the reconciliation of A completes
- * it should be applied (should update the assignment on the member and send ack), and then
- * the reconciliation of assignment B will be processed and applied in the next
- * reconciliation loop.
+ * This is the case where a member is stuck reconciling an assignment A (waiting on metadata, commit or callbacks), and the target
+ * assignment changes due to new topics. If the reconciliation of A completes it should be applied (should update the assignment
+ * on the member and send ack), and then the reconciliation of assignment B will be processed and applied in the next reconciliation
+ * loop.
*/
@Test
public void testDelayedReconciliationResultAppliedWhenTargetChangedWithMetadataUpdate() {
@@ -539,8 +536,8 @@ public class MembershipManagerImplTest {
// Metadata discovered for topic2 while reconciliation in progress to revoke topic1.
// Should not trigger a new reconciliation because there is one already in progress.
- mockTopicNameInMetadataCache(Collections.singletonMap(topicId2, topic2), true);
- membershipManager.onUpdate(null);
+ final Map<Uuid, String> topic2Metadata = Collections.singletonMap(topicId2, topic2);
+ mockTopicNameInMetadataCache(topic2Metadata, true);
assertEquals(Collections.singleton(topicId2), membershipManager.topicsAwaitingReconciliation());
verifyReconciliationNotTriggered(membershipManager);
@@ -557,6 +554,86 @@ public class MembershipManagerImplTest {
// next reconciliation loop.
Map<Uuid, SortedSet<Integer>> topic2Assignment = topicIdPartitionsMap(topicId2, 1, 2);
assertEquals(topic2Assignment, membershipManager.topicPartitionsAwaitingReconciliation());
+
+ // After acknowledging the assignment, we should be back to RECONCILING, because we have not
+ // yet reached the target assignment.
+ membershipManager.onHeartbeatRequestSent();
+
+ assertEquals(MemberState.RECONCILING, membershipManager.state());
+ assertEquals(topic2Assignment, membershipManager.topicPartitionsAwaitingReconciliation());
+
+ // Next reconciliation triggered in poll
+ membershipManager.poll(time.milliseconds());
+
+ assertEquals(Collections.emptySet(), membershipManager.topicsAwaitingReconciliation());
+ verify(subscriptionState).assignFromSubscribed(topicPartitions(topic2Assignment, topic2Metadata));
+ }
+
+ /**
+ * This is the case where a member is stuck reconciling an assignment A (waiting on metadata, commit or callbacks), and the target
+ * assignment changes due to a new assignment received from broker. If the reconciliation of A completes it should be applied (should
+ * update the assignment on the member and send ack), and then the reconciliation of assignment B will be processed and applied in the
+ * next reconciliation loop.
+ */
+ @Test
+ public void testDelayedReconciliationResultAppliedWhenTargetChangedWithNewAssignment() {
+
+ Uuid topicId1 = Uuid.randomUuid();
+ String topic1 = "topic1";
+ final TopicIdPartition topicId1Partition0 = new TopicIdPartition(topicId1, new TopicPartition(topic1, 0));
+
+ Uuid topicId2 = Uuid.randomUuid();
+ String topic2 = "topic2";
+ final TopicIdPartition topicId2Partition0 = new TopicIdPartition(topicId2, new TopicPartition(topic2, 0));
+
+ MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();
+ when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
+ when(metadata.topicNames()).thenReturn(
+ mkMap(
+ mkEntry(topicId1, topic1),
+ mkEntry(topicId2, topic2)
+ )
+ );
+
+ // Receive assignment with only topic1-0, getting stuck during commit.
+ final CompletableFuture<Void> commitFuture = mockNewAssignmentAndRevocationStuckOnCommit(membershipManager, topicId1,
+ topic1, Collections.singletonList(0), false);
+
+ // New assignment adding a new topic2-0 (not in metadata).
+ // No reconciliation triggered, because another reconciliation is in progress.
+ Map<Uuid, SortedSet<Integer>> newAssignment =
+ mkMap(
+ mkEntry(topicId1, mkSortedSet(0)),
+ mkEntry(topicId2, mkSortedSet(0))
+ );
+
+ receiveAssignment(newAssignment, membershipManager);
+ membershipManager.poll(time.milliseconds());
+
+ verifyReconciliationNotTriggered(membershipManager);
+ assertEquals(MemberState.RECONCILING, membershipManager.state());
+ assertEquals(mkSet(topicId1, topicId2), membershipManager.topicsAwaitingReconciliation());
+ clearInvocations(membershipManager, commitRequestManager);
+
+ // First reconciliation completes. Should trigger follow-up reconciliation to complete the assignment,
+ // with membership manager entering ACKNOWLEDGING state.
+
+ commitFuture.complete(null);
+
+ assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
+ assertEquals(mkSet(topicId2), membershipManager.topicsAwaitingReconciliation());
+
+ // After acknowledging the assignment, we should be back to RECONCILING, because we have not
+ // yet reached the target assignment.
+ membershipManager.onHeartbeatRequestSent();
+
+ assertEquals(MemberState.RECONCILING, membershipManager.state());
+ clearInvocations(membershipManager, commitRequestManager);
+
+ // Next poll should trigger final reconciliation
+ membershipManager.poll(time.milliseconds());
+
+ verifyReconciliationTriggeredAndCompleted(membershipManager, Arrays.asList(topicId1Partition0, topicId2Partition0));
}
// Tests the case where topic metadata is not available at the time of the assignment,
@@ -592,6 +669,7 @@ public class MembershipManagerImplTest {
);
receiveAssignment(newAssignment, membershipManager);
+ membershipManager.poll(time.milliseconds());
verifyReconciliationNotTriggered(membershipManager);
assertEquals(MemberState.RECONCILING, membershipManager.state());
@@ -608,7 +686,7 @@ public class MembershipManagerImplTest {
);
when(metadata.topicNames()).thenReturn(fullTopicMetadata);
- membershipManager.onUpdate(null);
+ membershipManager.poll(time.milliseconds());
verifyReconciliationTriggeredAndCompleted(membershipManager, Arrays.asList(topicId1Partition0, topicId2Partition0));
}
@@ -637,6 +715,9 @@ public class MembershipManagerImplTest {
receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
List<TopicIdPartition> assignedPartitions = Arrays.asList(
new TopicIdPartition(topicId, new TopicPartition(topicName, 0)),
new TopicIdPartition(topicId, new TopicPartition(topicName, 1)));
@@ -818,6 +899,10 @@ public class MembershipManagerImplTest {
String topicName = "topic1";
when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, topicName));
receiveAssignment(topicId, Collections.singletonList(0), membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
Set<TopicPartition> expectedAssignment = Collections.singleton(new TopicPartition(topicName, 0));
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
verify(subscriptionState).assignFromSubscribed(expectedAssignment);
@@ -855,6 +940,10 @@ public class MembershipManagerImplTest {
String topicName = "topic1";
when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, topicName));
receiveEmptyAssignment(membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.STABLE, membershipManager.state());
verify(subscriptionState, never()).assignFromSubscribed(any());
}
@@ -872,6 +961,10 @@ public class MembershipManagerImplTest {
Uuid topicId = Uuid.randomUuid();
when(metadata.topicNames()).thenReturn(Collections.emptyMap());
receiveAssignment(topicId, Collections.singletonList(0), membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertFalse(membershipManager.topicsAwaitingReconciliation().isEmpty());
assertEquals(topicId, membershipManager.topicsAwaitingReconciliation().iterator().next());
@@ -896,7 +989,8 @@ public class MembershipManagerImplTest {
when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, topicName));
when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
when(subscriptionState.rebalanceListener()).thenReturn(Optional.empty());
- membershipManager.onUpdate(null);
+
+ membershipManager.poll(time.milliseconds());
// Assignment should have been reconciled.
Set<TopicPartition> expectedAssignment = Collections.singleton(new TopicPartition(topicName, 1));
@@ -957,6 +1051,9 @@ public class MembershipManagerImplTest {
receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
List<TopicIdPartition> assignedPartitions = topicIdPartitions(topicId, topicName, 0, 1);
verifyReconciliationTriggeredAndCompleted(membershipManager, assignedPartitions);
}
@@ -973,6 +1070,9 @@ public class MembershipManagerImplTest {
// New assignment received, adding partitions 1 and 2 to the previously owned partition 0.
receiveAssignment(topicId, Arrays.asList(0, 1, 2), membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
List<TopicIdPartition> assignedPartitions = new ArrayList<>();
assignedPartitions.add(ownedPartition);
assignedPartitions.addAll(topicIdPartitions(topicId, topicName, 1, 2));
@@ -990,6 +1090,10 @@ public class MembershipManagerImplTest {
mockOwnedPartitionAndAssignmentReceived(membershipManager, topicId, topicName, Collections.emptyList());
List<TopicIdPartition> expectedAssignmentReconciled = topicIdPartitions(topicId, topicName, 0, 1);
receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
verifyReconciliationTriggeredAndCompleted(membershipManager, expectedAssignmentReconciled);
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
clearInvocations(subscriptionState, membershipManager);
@@ -1017,6 +1121,9 @@ public class MembershipManagerImplTest {
receiveEmptyAssignment(membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
testRevocationOfAllPartitionsCompleted(membershipManager);
}
@@ -1029,6 +1136,9 @@ public class MembershipManagerImplTest {
receiveEmptyAssignment(membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
// Member stays in RECONCILING while the commit request hasn't completed.
assertEquals(MemberState.RECONCILING, membershipManager.state());
@@ -1050,6 +1160,9 @@ public class MembershipManagerImplTest {
receiveEmptyAssignment(membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
// Member stays in RECONCILING while the commit request hasn't completed.
assertEquals(MemberState.RECONCILING, membershipManager.state());
// Partitions should be still owned by the member
@@ -1077,6 +1190,9 @@ public class MembershipManagerImplTest {
// New assignment received, revoking partition 0, and assigning new partitions 1 and 2.
receiveAssignment(topicId, Arrays.asList(1, 2), membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
assertEquals(topicIdPartitionsMap(topicId, 1, 2), membershipManager.currentAssignment());
assertFalse(membershipManager.reconciliationInProgress());
@@ -1105,8 +1221,8 @@ public class MembershipManagerImplTest {
String topicName = "topic1";
mockTopicNameInMetadataCache(Collections.singletonMap(topicId, topicName), true);
- // When metadata is updated, the member should re-trigger reconciliation
- membershipManager.onUpdate(null);
+ // When the next poll is run, the member should re-trigger reconciliation
+ membershipManager.poll(time.milliseconds());
List<TopicIdPartition> expectedAssignmentReconciled = topicIdPartitions(topicId, topicName, 0, 1);
verifyReconciliationTriggeredAndCompleted(membershipManager, expectedAssignmentReconciled);
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
@@ -1131,10 +1247,10 @@ public class MembershipManagerImplTest {
assertEquals(Collections.singleton(topicId), membershipManager.topicsAwaitingReconciliation());
verify(metadata).requestUpdate(anyBoolean());
- // Metadata update received, but still without the unresolved topic in it. Should keep
+ // Next poll is run, but metadata still without the unresolved topic in it. Should keep
// the unresolved and request update again.
when(metadata.topicNames()).thenReturn(Collections.emptyMap());
- membershipManager.onUpdate(null);
+ membershipManager.poll(time.milliseconds());
verifyReconciliationNotTriggered(membershipManager);
assertEquals(Collections.singleton(topicId), membershipManager.topicsAwaitingReconciliation());
verify(metadata, times(2)).requestUpdate(anyBoolean());
@@ -1152,6 +1268,9 @@ public class MembershipManagerImplTest {
receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
// Member should complete reconciliation
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
List<Integer> partitions = Arrays.asList(0, 1);
@@ -1173,6 +1292,8 @@ public class MembershipManagerImplTest {
// Revoke one of the 2 partitions
receiveAssignment(topicId, Collections.singletonList(1), membershipManager);
+ membershipManager.poll(time.milliseconds());
+
// Revocation should complete without requesting any metadata update given that the topic
// received in target assignment should exist in local topic name cache.
verify(metadata, never()).requestUpdate(anyBoolean());
@@ -1209,6 +1330,10 @@ public class MembershipManagerImplTest {
// Step 2: put the state machine into the appropriate... state
when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, topicName));
receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertTrue(membershipManager.reconciliationInProgress());
assertEquals(0, listener.revokedCount());
@@ -1240,6 +1365,8 @@ public class MembershipManagerImplTest {
// Step 5: receive an empty assignment, which means we should call revoke
when(subscriptionState.assignedPartitions()).thenReturn(topicPartitions(topicName, 0, 1));
receiveEmptyAssignment(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertTrue(membershipManager.reconciliationInProgress());
@@ -1293,6 +1420,10 @@ public class MembershipManagerImplTest {
// Step 2: put the state machine into the appropriate... state
receiveEmptyAssignment(membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertEquals(topicIdPartitionsMap(topicId, 0), membershipManager.currentAssignment());
assertTrue(membershipManager.reconciliationInProgress());
@@ -1342,6 +1473,10 @@ public class MembershipManagerImplTest {
// Step 2: put the state machine into the appropriate... state
receiveEmptyAssignment(membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertEquals(topicIdPartitionsMap(topicId, 0), membershipManager.currentAssignment());
assertTrue(membershipManager.reconciliationInProgress());
@@ -1396,6 +1531,8 @@ public class MembershipManagerImplTest {
mockPartitionOwnedAndNewPartitionAdded(topicName, partitionOwned, partitionAdded,
new CounterConsumerRebalanceListener(), membershipManager);
+ membershipManager.poll(time.milliseconds());
+
verify(subscriptionState).assignFromSubscribedAwaitingCallback(assignedPartitions, addedPartitions);
performCallback(
@@ -1426,6 +1563,8 @@ public class MembershipManagerImplTest {
mockPartitionOwnedAndNewPartitionAdded(topicName, partitionOwned, partitionAdded,
listener, membershipManager);
+ membershipManager.poll(time.milliseconds());
+
verify(subscriptionState).assignFromSubscribedAwaitingCallback(assignedPartitions, addedPartitions);
performCallback(
@@ -1521,6 +1660,18 @@ public class MembershipManagerImplTest {
);
}
+ private static SortedSet<TopicPartition> topicPartitions(Map<Uuid, SortedSet<Integer>> topicIdMap, Map<Uuid, String> topicIdNames) {
+ SortedSet<TopicPartition> topicPartitions = new TreeSet<>(new Utils.TopicPartitionComparator());
+
+ for (Uuid topicId : topicIdMap.keySet()) {
+ for (int partition : topicIdMap.get(topicId)) {
+ topicPartitions.add(new TopicPartition(topicIdNames.get(topicId), partition));
+ }
+ }
+
+ return topicPartitions;
+ }
+
private SortedSet<TopicPartition> topicPartitions(String topicName, int... partitions) {
SortedSet<TopicPartition> topicPartitions = new TreeSet<>(new Utils.TopicPartitionComparator());
@@ -1613,6 +1764,10 @@ public class MembershipManagerImplTest {
MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup(null);
assertEquals(MemberState.JOINING, membershipManager.state());
receiveEmptyAssignment(membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
assertEquals(MemberState.STABLE, membershipManager.state());
}
@@ -1624,6 +1779,9 @@ public class MembershipManagerImplTest {
receiveAssignment(topicId, partitions, membershipManager);
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
List<TopicIdPartition> assignedPartitions =
partitions.stream().map(tp -> new TopicIdPartition(topicId,
new TopicPartition(topicName, tp))).collect(Collectors.toList());
@@ -1635,6 +1793,10 @@ public class MembershipManagerImplTest {
MembershipManagerImpl membershipManager) {
CompletableFuture<Void> commitResult = mockRevocationNoCallbacks(true);
receiveEmptyAssignment(membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
verifyReconciliationTriggered(membershipManager);
clearInvocations(membershipManager);
assertEquals(MemberState.RECONCILING, membershipManager.state());
@@ -1650,6 +1812,10 @@ public class MembershipManagerImplTest {
when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, topicName));
}
receiveAssignment(topicId, partitions, membershipManager);
+
+ verifyReconciliationNotTriggered(membershipManager);
+ membershipManager.poll(time.milliseconds());
+
verifyReconciliationTriggered(membershipManager);
clearInvocations(membershipManager);
assertEquals(MemberState.RECONCILING, membershipManager.state());
@@ -1784,6 +1950,8 @@ public class MembershipManagerImplTest {
when(subscriptionState.rebalanceListener()).thenReturn(Optional.empty()).thenReturn(Optional.empty());
membershipManager.onHeartbeatResponseReceived(heartbeatResponse.data());
+ membershipManager.poll(time.milliseconds());
+
if (expectSubscriptionUpdated) {
verify(subscriptionState).assignFromSubscribed(anyCollection());
} else {
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
index 075552fcd55..8ea8cb7a729 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
@@ -77,7 +77,9 @@ public class ApplicationEventProcessorTest {
fetchRequestManager,
Optional.of(coordinatorRequestManager),
Optional.of(commitRequestManager),
- Optional.of(heartbeatRequestManager));
+ Optional.of(heartbeatRequestManager),
+ Optional.of(membershipManager)
+ );
processor = new ApplicationEventProcessor(
new LogContext(),
applicationEventQueue,