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/06/14 11:04:03 UTC
(kafka) branch trunk updated: KAFKA-16933: New consumer unsubscribe close commit fixes (#16272)
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 46714dbaed0 KAFKA-16933: New consumer unsubscribe close commit fixes (#16272)
46714dbaed0 is described below
commit 46714dbaed066e8d482d4976259bdc500c97cc47
Author: Lianet Magrans <98...@users.noreply.github.com>
AuthorDate: Fri Jun 14 13:03:58 2024 +0200
KAFKA-16933: New consumer unsubscribe close commit fixes (#16272)
Fixes for the leave group flow (unsubscribe/close):
Fix to send Heartbeat to leave group on close even if the callbacks fail
fix to ensure that if a member gets fenced while blocked on callbacks (ex. on unsubscribe), it will clear its epoch to not include it in commit requests
fix to avoid race on the subscription state object on unsubscribe, updating it only on the background thread when the callbacks to leave complete (success or failure).
Also improving logging in this area.
Reviewers: Lucas Brutschy <lb...@confluent.io>, Philip Nee <pn...@confluent.io>
---
.../consumer/internals/AsyncKafkaConsumer.java | 19 ++++----
.../internals/HeartbeatRequestManager.java | 3 +-
.../consumer/internals/MembershipManagerImpl.java | 53 ++++++++++++++++------
.../kafka/clients/consumer/KafkaConsumerTest.java | 2 +-
.../consumer/internals/AsyncKafkaConsumerTest.java | 13 ++++--
.../internals/MembershipManagerImplTest.java | 2 +
6 files changed, 64 insertions(+), 28 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 4a4012f4ddd..384bce0f670 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -1278,12 +1278,10 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
commitSyncAllConsumed(timer);
applicationEventHandler.add(new CommitOnCloseEvent());
- completeQuietly(
- () -> {
- maybeRevokePartitions();
- applicationEventHandler.addAndGet(new LeaveOnCloseEvent(calculateDeadlineMs(timer)));
- },
- "Failed to send leaveGroup heartbeat with a timeout(ms)=" + timer.timeoutMs(), firstException);
+ completeQuietly(() -> maybeRevokePartitions(),
+ "Failed to execute callback to release assignment", firstException);
+ completeQuietly(() -> applicationEventHandler.addAndGet(new LeaveOnCloseEvent(calculateDeadlineMs(timer))),
+ "Failed to send leave group heartbeat with a timeout(ms)=" + timer.timeoutMs(), firstException);
}
// Visible for testing
@@ -1323,7 +1321,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
function.run();
} catch (TimeoutException e) {
log.debug("Timeout expired before the {} operation could complete.", msg);
+ firstException.compareAndSet(null, e);
} catch (Exception e) {
+ log.error(msg, e);
firstException.compareAndSet(null, e);
}
}
@@ -1502,7 +1502,8 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
Timer timer = time.timer(Long.MAX_VALUE);
UnsubscribeEvent unsubscribeEvent = new UnsubscribeEvent(calculateDeadlineMs(timer));
applicationEventHandler.add(unsubscribeEvent);
- log.info("Unsubscribing all topics or patterns and assigned partitions");
+ log.info("Unsubscribing all topics or patterns and assigned partitions {}",
+ subscriptions.assignedPartitions());
try {
processBackgroundEvents(unsubscribeEvent.future(), timer);
@@ -1512,7 +1513,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
}
resetGroupMetadata();
}
- subscriptions.unsubscribe();
+ } catch (Exception e) {
+ log.error("Unsubscribe failed", e);
+ throw e;
} finally {
release();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
index 0ed4a67b586..fdfd662ea41 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
@@ -263,7 +263,8 @@ public class HeartbeatRequestManager implements RequestManager {
pollTimer.update(pollMs);
if (pollTimer.isExpired()) {
logger.warn("Time between subsequent calls to poll() was longer than the configured " +
- "max.poll.interval.ms, exceeded approximately by {} ms.", pollTimer.isExpiredBy());
+ "max.poll.interval.ms, exceeded approximately by {} ms. Member {} will rejoin the group now.",
+ pollTimer.isExpiredBy(), membershipManager.memberId());
membershipManager.maybeRejoinStaleMember();
}
pollTimer.reset(maxPollIntervalMs);
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 0e894a2a401..10390fbb185 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
@@ -46,6 +46,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -520,12 +521,15 @@ public class MembershipManagerImpl implements MembershipManager {
@Override
public void transitionToFenced() {
if (state == MemberState.PREPARE_LEAVING) {
- log.debug("Member {} with epoch {} got fenced but it is already preparing to leave " +
+ log.info("Member {} with epoch {} got fenced but it is already preparing to leave " +
"the group, so it will stop sending heartbeat and won't attempt to rejoin.",
memberId, memberEpoch);
- // Transition to UNSUBSCRIBED, ensuring that the member (that is not part of the
- // group anymore from the broker point of view) will stop sending heartbeats while it
- // completes the ongoing leaving operation.
+ // Briefly transition to LEAVING to ensure all required actions are applied even
+ // though there is no need to send a leave group heartbeat (ex. clear epoch and
+ // notify epoch listeners). Then transition to UNSUBSCRIBED, ensuring that the member
+ // (that is not part of the group anymore from the broker point of view) will stop
+ // sending heartbeats while it completes the ongoing leaving operation.
+ transitionToSendingLeaveGroup(false);
transitionTo(MemberState.UNSUBSCRIBED);
return;
}
@@ -552,7 +556,7 @@ public class MembershipManagerImpl implements MembershipManager {
log.error("onPartitionsLost callback invocation failed while releasing assignment" +
" after member got fenced. Member will rejoin the group anyways.", error);
}
- clearSubscription();
+ clearAssignment();
if (state == MemberState.FENCED) {
transitionToJoining();
} else {
@@ -589,7 +593,7 @@ public class MembershipManagerImpl implements MembershipManager {
log.error("onPartitionsLost callback invocation failed while releasing assignment" +
"after member failed with fatal error.", error);
}
- clearSubscription();
+ clearAssignment();
});
}
@@ -605,7 +609,7 @@ public class MembershipManagerImpl implements MembershipManager {
/**
* Clear the assigned partitions in the member subscription, pending assignments and metadata cache.
*/
- private void clearSubscription() {
+ private void clearAssignment() {
if (subscriptions.hasAutoAssignedPartitions()) {
subscriptions.assignFromSubscribed(Collections.emptySet());
}
@@ -655,7 +659,7 @@ public class MembershipManagerImpl implements MembershipManager {
public CompletableFuture<Void> leaveGroup() {
if (isNotInGroup()) {
if (state == MemberState.FENCED) {
- clearSubscription();
+ clearAssignment();
transitionTo(MemberState.UNSUBSCRIBED);
}
return CompletableFuture.completedFuture(null);
@@ -664,6 +668,7 @@ public class MembershipManagerImpl implements MembershipManager {
if (state == MemberState.PREPARE_LEAVING || state == MemberState.LEAVING) {
// Member already leaving. No-op and return existing leave group future that will
// complete when the ongoing leave operation completes.
+ log.debug("Leave group operation already in progress for member {}", memberId);
return leaveGroupInProgress.get();
}
@@ -673,8 +678,16 @@ public class MembershipManagerImpl implements MembershipManager {
CompletableFuture<Void> callbackResult = invokeOnPartitionsRevokedOrLostToReleaseAssignment();
callbackResult.whenComplete((result, error) -> {
+ if (error != null) {
+ log.error("Member {} callback to release assignment failed. Member will proceed " +
+ "to send leave group heartbeat", memberId, error);
+ } else {
+ log.debug("Member {} completed callback to release assignment and will send leave " +
+ "group heartbeat", memberId);
+ }
// Clear the subscription, no matter if the callback execution failed or succeeded.
- clearSubscription();
+ subscriptions.unsubscribe();
+ clearAssignment();
// Transition to ensure that a heartbeat request is sent out to effectively leave the
// group (even in the case where the member had no assignment to release or when the
@@ -705,6 +718,9 @@ public class MembershipManagerImpl implements MembershipManager {
SortedSet<TopicPartition> droppedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
droppedPartitions.addAll(subscriptions.assignedPartitions());
+ log.debug("Member {} is triggering callbacks to release assignment {} and leave group",
+ memberId, droppedPartitions);
+
CompletableFuture<Void> callbackResult;
if (droppedPartitions.isEmpty()) {
// No assignment to release.
@@ -764,7 +780,7 @@ public class MembershipManagerImpl implements MembershipManager {
* This also includes the latest member ID in the notification. If the member fails or leaves
* the group, this will be invoked with empty epoch and member ID.
*/
- private void notifyEpochChange(Optional<Integer> epoch, Optional<String> memberId) {
+ void notifyEpochChange(Optional<Integer> epoch, Optional<String> memberId) {
stateUpdatesListeners.forEach(stateListener -> stateListener.onMemberEpochUpdated(epoch, memberId));
}
@@ -794,8 +810,12 @@ public class MembershipManagerImpl implements MembershipManager {
}
} else if (state == MemberState.LEAVING) {
if (isPollTimerExpired) {
+ log.debug("Member {} sent heartbeat to leave due to expired poll timer. It will " +
+ "remain stale (no heartbeat) until it rejoins the group on the next consumer " +
+ "poll.", memberId);
transitionToStale();
} else {
+ log.debug("Member {} sent heartbeat to leave group.", memberId);
transitionToUnsubscribed();
}
}
@@ -875,7 +895,7 @@ public class MembershipManagerImpl implements MembershipManager {
log.error("onPartitionsLost callback invocation failed while releasing assignment" +
" after member left group due to expired poll timer.", error);
}
- clearSubscription();
+ clearAssignment();
log.debug("Member {} sent leave group heartbeat and released its assignment. It will remain " +
"in {} state until the poll timer is reset, and it will then rejoin the group",
memberId, MemberState.STALE);
@@ -939,11 +959,13 @@ public class MembershipManagerImpl implements MembershipManager {
revokedPartitions.addAll(ownedPartitions);
revokedPartitions.removeAll(assignedTopicPartitions);
- log.info("Updating assignment with local epoch {}\n" +
+ log.info("Reconciling assignment with local epoch {}\n" +
+ "\tMember: {}\n" +
"\tAssigned partitions: {}\n" +
"\tCurrent owned partitions: {}\n" +
"\tAdded partitions (assigned - owned): {}\n" +
"\tRevoked partitions (owned - assigned): {}\n",
+ memberId,
resolvedAssignment.localEpoch,
assignedTopicPartitions,
ownedPartitions,
@@ -1175,11 +1197,14 @@ public class MembershipManagerImpl implements MembershipManager {
* request fails, this will proceed to invoke the user callbacks anyway,
* returning a future that will complete or fail depending on the callback execution only.
*
- * @param revokedPartitions Partitions to revoke.
+ * @param partitionsToRevoke Partitions to revoke.
* @return Future that will complete when the commit request and user callback completes.
* Visible for testing
*/
- CompletableFuture<Void> revokePartitions(Set<TopicPartition> revokedPartitions) {
+ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke) {
+ // Ensure the set of partitions to revoke are still assigned
+ Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
+ revokedPartitions.retainAll(subscriptions.assignedPartitions());
log.info("Revoking previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
logPausedPartitionsBeingRevoked(revokedPartitions);
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index d34d09cd8a9..cfe4d03e6a8 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -435,7 +435,7 @@ public class KafkaConsumerTest {
}
@ParameterizedTest
- @EnumSource(GroupProtocol.class)
+ @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
public void testSubscription(GroupProtocol groupProtocol) {
consumer = newConsumer(groupProtocol, groupId);
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index 2d126b3eac6..6a4cae65706 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -158,7 +158,12 @@ public class AsyncKafkaConsumerTest {
public void resetAll() {
backgroundEventQueue.clear();
if (consumer != null) {
- consumer.close(Duration.ZERO);
+ try {
+ consumer.close(Duration.ZERO);
+ } catch (Exception e) {
+ // best effort to clean up after each test, but may throw (ex. if callbacks where
+ // throwing errors)
+ }
}
consumer = null;
Mockito.framework().clearInlineMocks();
@@ -832,8 +837,8 @@ public class AsyncKafkaConsumerTest {
@Test
public void testFailedPartitionRevocationOnClose() {
- // If rebalance listener failed to execute during close, we will skip sending leave group and proceed with
- // closing the consumer.
+ // If rebalance listener failed to execute during close, we still send the leave group,
+ // and proceed with closing the consumer.
ConsumerRebalanceListener listener = mock(ConsumerRebalanceListener.class);
SubscriptionState subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
consumer = newConsumer(
@@ -848,7 +853,7 @@ public class AsyncKafkaConsumerTest {
subscriptions.assignFromSubscribed(singleton(tp));
doThrow(new KafkaException()).when(listener).onPartitionsRevoked(eq(singleton(tp)));
assertThrows(KafkaException.class, () -> consumer.close(Duration.ZERO));
- verify(applicationEventHandler, never()).addAndGet(any(LeaveOnCloseEvent.class));
+ verify(applicationEventHandler).addAndGet(any(LeaveOnCloseEvent.class));
verify(listener).onPartitionsRevoked(eq(singleton(tp)));
assertEquals(emptySet(), subscriptions.assignedPartitions());
}
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 cef608a902d..44de312087a 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
@@ -365,6 +365,8 @@ public class MembershipManagerImplTest {
// because member is already out of the group in the broker).
completeCallback(callbackEvent, membershipManager);
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ assertEquals(ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH, membershipManager.memberEpoch());
+ verify(membershipManager).notifyEpochChange(Optional.empty(), Optional.empty());
assertTrue(membershipManager.shouldSkipHeartbeat());
}