You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2022/04/13 11:47:57 UTC
[kafka] branch 3.2 updated: KAFKA-13828; Ensure reasons sent by the consumer are small (#12043)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.2 by this push:
new 4e7cc335ee KAFKA-13828; Ensure reasons sent by the consumer are small (#12043)
4e7cc335ee is described below
commit 4e7cc335ee007c90d8fc7b3b61f42b1de81f98af
Author: David Jacot <dj...@confluent.io>
AuthorDate: Wed Apr 13 13:42:27 2022 +0200
KAFKA-13828; Ensure reasons sent by the consumer are small (#12043)
This PR reworks the reasons used in the ConsumerCoordinator to ensure that they remain reasonably short.
Reviewers: Bruno Cadonna <br...@confluent.io>
---
.../consumer/internals/AbstractCoordinator.java | 23 +++++++++++++++++-----
.../consumer/internals/ConsumerCoordinator.java | 16 +++++++--------
2 files changed, 26 insertions(+), 13 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index 5b9712f346..b2f944ad5d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -1022,15 +1022,28 @@ public abstract class AbstractCoordinator implements Closeable {
resetStateAndRejoin("consumer pro-actively leaving the group", true);
}
- public synchronized void requestRejoinIfNecessary(final String reason) {
+ public synchronized void requestRejoinIfNecessary(final String shortReason,
+ final String fullReason) {
if (!this.rejoinNeeded) {
- requestRejoin(reason);
+ requestRejoin(shortReason, fullReason);
}
}
- public synchronized void requestRejoin(final String reason) {
- log.info("Request joining group due to: {}", reason);
- this.rejoinReason = reason;
+ public synchronized void requestRejoin(final String shortReason) {
+ requestRejoin(shortReason, shortReason);
+ }
+
+ /**
+ * Request to rejoin the group.
+ *
+ * @param shortReason This is the reason passed up to the group coordinator. It must be
+ * reasonably small.
+ * @param fullReason This is the reason logged locally.
+ */
+ public synchronized void requestRejoin(final String shortReason,
+ final String fullReason) {
+ log.info("Request joining group due to: {}", fullReason);
+ this.rejoinReason = shortReason;
this.rejoinNeeded = true;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index 10939b2a0e..51fa0b62ed 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -401,10 +401,10 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
assignedPartitions.addAll(assignment.partitions());
if (!subscriptions.checkAssignmentMatchedSubscription(assignedPartitions)) {
- final String reason = String.format("received assignment %s does not match the current subscription %s; " +
+ final String fullReason = String.format("received assignment %s does not match the current subscription %s; " +
"it is likely that the subscription has changed since we joined the group, will re-join with current subscription",
assignment.partitions(), subscriptions.prettyString());
- requestRejoin(reason);
+ requestRejoin("received assignment does not match the current subscription", fullReason);
return;
}
@@ -437,9 +437,9 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
firstException.compareAndSet(null, invokePartitionsRevoked(revokedPartitions));
// If revoked any partitions, need to re-join the group afterwards
- final String reason = String.format("need to revoke partitions %s as indicated " +
+ final String fullReason = String.format("need to revoke partitions %s as indicated " +
"by the current assignment and re-join", revokedPartitions);
- requestRejoin(reason);
+ requestRejoin("need to revoke partitions and re-join", fullReason);
}
}
@@ -851,17 +851,17 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
// we need to rejoin if we performed the assignment and metadata has changed;
// also for those owned-but-no-longer-existed partitions we should drop them as lost
if (assignmentSnapshot != null && !assignmentSnapshot.matches(metadataSnapshot)) {
- final String reason = String.format("cached metadata has changed from %s at the beginning of the rebalance to %s",
+ final String fullReason = String.format("cached metadata has changed from %s at the beginning of the rebalance to %s",
assignmentSnapshot, metadataSnapshot);
- requestRejoinIfNecessary(reason);
+ requestRejoinIfNecessary("cached metadata has changed", fullReason);
return true;
}
// we need to join if our subscription has changed since the last join
if (joinedSubscription != null && !joinedSubscription.equals(subscriptions.subscription())) {
- final String reason = String.format("subscription has changed from %s at the beginning of the rebalance to %s",
+ final String fullReason = String.format("subscription has changed from %s at the beginning of the rebalance to %s",
joinedSubscription, subscriptions.subscription());
- requestRejoinIfNecessary(reason);
+ requestRejoinIfNecessary("subscription has changed", fullReason);
return true;
}