You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "philipnee (via GitHub)" <gi...@apache.org> on 2024/02/08 05:18:59 UTC
[PR] KAFKA-16116 [kafka]
philipnee opened a new pull request, #15339:
URL: https://github.com/apache/kafka/pull/15339
Adding the following rebalance metrics to the consumer:
rebalance-latency-avg
rebalance-latency-max
rebalance-latency-total
rebalance-rate-per-hour
rebalance-total
failed-rebalance-rate-per-hour
failed-rebalance-total
Due to the difference in protocol, we need to redefine when rebalance starts and ends.
**Start of Rebalance:**
Current: Right before sending out JoinGroup
ConsumerGroup: When the client receives assignments from the HB
**End of Rebalance - Successful Case:**
Current: Receiving SyncGroup request after transitioning to "COMPLETING_REBALANCE"
ConsumerGroup: After completing reconciliation and right before sending out "Ack" heartbeat
**End of Rebalance - Failed Case:**
Current: Any failure in the JoinGroup/SyncGroup response
ConsumerGroup: Failure in the heartbeat
Note: Afterall, we try to be consistent with the current protocol. Rebalances start and end with sending and receiving network requests. Failures in network requests signify the user failures in rebalance. And it is entirely possible to have multiple failures before having a successful one.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16116: Rebalance Metrics for AsyncKafkaConsumer [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #15339:
URL: https://github.com/apache/kafka/pull/15339#issuecomment-1934623723
Hi @lucasbru - Would you have time to review this PR? The desired logic is written in the description. Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16116: Rebalance Metrics for AsyncKafkaConsumer [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15339:
URL: https://github.com/apache/kafka/pull/15339#discussion_r1485988931
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -95,6 +98,8 @@ public class MembershipManagerImplTest {
private BlockingQueue<BackgroundEvent> backgroundEventQueue;
private BackgroundEventHandler backgroundEventHandler;
private Time time;
+ private Metrics metrics;
+ private RebalanceMetricsManager rebalanceMetricsManager = mock(RebalanceMetricsManager.class);
Review Comment:
In this class, you sometimes mock the rebalancemetricsmanager, and sometimes you use the real one. I think it's easier to stick to one of the two, so either the metrics manager is part of the unit-under-test or not. You can probably just use the "real one" everywhere?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -242,7 +234,7 @@ public void testTransitionToFailedWhenTryingToJoin() {
MembershipManagerImpl membershipManager = new MembershipManagerImpl(
GROUP_ID, Optional.empty(), REBALANCE_TIMEOUT, Optional.empty(),
subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
- backgroundEventHandler, time);
+ backgroundEventHandler, time, mock(RebalanceMetricsManager.class));
Review Comment:
why no use your member `rebalanceMetricsManager` here?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -163,7 +155,7 @@ public void testMembershipManagerRegistersForClusterMetadataUpdatesOnFirstJoin()
MembershipManagerImpl manager = new MembershipManagerImpl(
GROUP_ID, Optional.empty(), REBALANCE_TIMEOUT, Optional.empty(),
subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
- backgroundEventHandler, time);
+ backgroundEventHandler, time, mock(RebalanceMetricsManager.class));
Review Comment:
why no use your member `rebalanceMetricsManager` here?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java:
##########
@@ -75,6 +75,11 @@ public interface MembershipManager {
*/
void onHeartbeatResponseReceived(ConsumerGroupHeartbeatResponseData response);
+ /**
+ * Notify the member that an error heartbeat response was received.
+ */
+ void onHeartbeatError();
Review Comment:
I wonder if it would make sense to rename `onHeartbeatResponseReceived` in this change, since an error response is also a response that we receive -- it should be clear that it's only called in the successful case. Maybe we could call the two functions `onHeartbeatSuccess` and `onHeartbeatFailure`, going along with the convention in `RequestFutureListener` etc.pp.?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -1610,12 +1605,152 @@ public void testHeartbeatSentOnStaledMember() {
@Test
public void testMemberJoiningTransitionsToStableWhenReceivingEmptyAssignment() {
- MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup(null);
+ MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();
assertEquals(MemberState.JOINING, membershipManager.state());
receiveEmptyAssignment(membershipManager);
assertEquals(MemberState.STABLE, membershipManager.state());
}
+ @Test
+ public void testMetricsWhenHeartbeatFailed() {
+ rebalanceMetricsManager = new RebalanceMetricsManager(metrics);
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+ membershipManager.onHeartbeatError();
+
+ // Not expecting rebalance failures without assignments being reconciled
+ assertEquals(0.0d, getMetricValue(metrics, rebalanceMetricsManager.rebalanceTotal));
+ assertEquals(0.0d, getMetricValue(metrics, rebalanceMetricsManager.failedRebalanceTotal));
+ }
+
+ @Test
+ public void testRebalanceMetricsOnSuccessfulRebalance() {
+ rebalanceMetricsManager = new RebalanceMetricsManager(metrics);
+ MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();
+ ConsumerGroupHeartbeatResponse heartbeatResponse = createConsumerGroupHeartbeatResponse(null);
+ membershipManager.onHeartbeatResponseReceived(heartbeatResponse.data());
+ mockOwnedPartition(membershipManager, Uuid.randomUuid(), "topic1");
+
+ CompletableFuture<Void> commitResult = mockRevocationNoCallbacks(true);
+
+ receiveEmptyAssignment(membershipManager);
+ long reconciliationDurationMs = sleepRandomMs(time, 2000);
+
+ // Complete commit request to complete the callback invocation
+ commitResult.complete(null);
+
+ assertEquals((double) reconciliationDurationMs, getMetricValue(metrics, rebalanceMetricsManager.rebalanceLatencyTotal));
+ assertEquals((double) reconciliationDurationMs, getMetricValue(metrics, rebalanceMetricsManager.rebalanceLatencyAvg));
+ assertEquals((double) reconciliationDurationMs, getMetricValue(metrics, rebalanceMetricsManager.rebalanceLatencyMax));
+ assertEquals(1d, getMetricValue(metrics, rebalanceMetricsManager.rebalanceTotal));
+ assertEquals(120d, 1d, (double) getMetricValue(metrics, rebalanceMetricsManager.rebalanceRatePerHour));
+ assertEquals(0d, getMetricValue(metrics, rebalanceMetricsManager.failedRebalanceRate));
+ assertEquals(0d, getMetricValue(metrics, rebalanceMetricsManager.failedRebalanceTotal));
+ assertEquals(0d, getMetricValue(metrics, rebalanceMetricsManager.lastRebalanceSecondsAgo));
+ }
+
+ @Test
+ public void testRebalanceMetricsForMultipleReconcilations() {
+ rebalanceMetricsManager = new RebalanceMetricsManager(metrics);
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+ ConsumerRebalanceListenerInvoker invoker = consumerRebalanceListenerInvoker();
+
+ String topicName = "topic1";
+ Uuid topicId = Uuid.randomUuid();
+
+ SleepyRebalanceListener listener = new SleepyRebalanceListener(2000, time);
+ when(subscriptionState.assignedPartitions()).thenReturn(Collections.emptySet());
+ when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
+ when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(listener));
+ doNothing().when(subscriptionState).markPendingRevocation(anySet());
+ when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, topicName));
+
+ when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, topicName));
+ receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);
+
+ // assign partitions
+ performCallback(
+ membershipManager,
+ invoker,
+ ConsumerRebalanceListenerMethodName.ON_PARTITIONS_ASSIGNED,
+ topicPartitions(topicName, 0, 1),
+ true
+ );
+
+ long firstRebalanaceTimesMs = listener.sleepMs;
+ listener.reset();
+
+ // ack
+ membershipManager.onHeartbeatRequestSent();
+
+ // revoke all
+ when(subscriptionState.assignedPartitions()).thenReturn(topicPartitions(topicName, 0, 1));
+ receiveAssignment(topicId, Collections.singletonList(2), membershipManager);
+ performCallback(
+ membershipManager,
+ invoker,
+ ConsumerRebalanceListenerMethodName.ON_PARTITIONS_REVOKED,
+ topicPartitions(topicName, 0, 1),
+ true
+ );
+
+ // assign new partition 2
+ performCallback(
+ membershipManager,
+ invoker,
+ ConsumerRebalanceListenerMethodName.ON_PARTITIONS_ASSIGNED,
+ topicPartitions(topicName, 2),
+ true
+ );
+ membershipManager.onHeartbeatRequestSent();
+
+ long secondRebalanceMs = listener.sleepMs;
+ long total = firstRebalanaceTimesMs + secondRebalanceMs;
+ double avg = total / 2.0d;
+ long max = Math.max(firstRebalanaceTimesMs, secondRebalanceMs);
+ assertEquals((double) total, getMetricValue(metrics, rebalanceMetricsManager.rebalanceLatencyTotal));
+ assertEquals(avg, (double) getMetricValue(metrics, rebalanceMetricsManager.rebalanceLatencyAvg), 1d);
+ assertEquals((double) max, getMetricValue(metrics, rebalanceMetricsManager.rebalanceLatencyMax));
+ assertEquals(2d, getMetricValue(metrics, rebalanceMetricsManager.rebalanceTotal));
+ // rate is not tested because it is subject to Rate implementation
+ assertEquals(0d, getMetricValue(metrics, rebalanceMetricsManager.failedRebalanceRate));
+ assertEquals(0d, getMetricValue(metrics, rebalanceMetricsManager.failedRebalanceTotal));
+ assertEquals(0d, getMetricValue(metrics, rebalanceMetricsManager.lastRebalanceSecondsAgo));
+
+ }
+
+ @Test
+ public void testRebalanceMetricsOnFailedRebalance() {
+ rebalanceMetricsManager = new RebalanceMetricsManager(metrics);
+ MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();
+ ConsumerGroupHeartbeatResponse heartbeatResponse = createConsumerGroupHeartbeatResponse(null);
+ membershipManager.onHeartbeatResponseReceived(heartbeatResponse.data());
+
+ Uuid topicId = Uuid.randomUuid();
+
+ receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);
+
+ sleepRandomMs(time, 2000);
Review Comment:
I think it's good practice to avoid randomness in unit tests, to make the tests repeatable. Otherwise, the confusing test results are typically not worth the extra coverage in my experience. If you want to use arbitrary numbers, maybe fix the seed of the PRNG. See, for example, how that Jay guy used a `SEEDED_RANDOM` in `TestUtils.randomBytes`. But I don't think Kafka is super consistent here, so it's more a suggestion, I won't insist.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -923,6 +965,7 @@ private void revokeAndAssign(SortedSet<TopicIdPartition> assignedTopicIdPartitio
} else {
boolean memberHasRejoined = memberEpochOnReconciliationStart != memberEpoch;
if (state == MemberState.RECONCILING && !memberHasRejoined) {
+ metricsManager.recordRebalanceEnded(time.milliseconds());
Review Comment:
I think there is at least one more situation where we finish a rebalance, which is when we transition from `RECONCILING` to `STABLE` directly when an unreconciled partition is removed from the assignment by the broker. In general, wouldn't it be simpler to just handle this inside `transititionTo`, start the measurement when we enter `RECONCILING` and stop the measurement when we leave `RECONCILING`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16116: Rebalance Metrics for AsyncKafkaConsumer [kafka]
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #15339:
URL: https://github.com/apache/kafka/pull/15339#issuecomment-1968168522
Hi @lucasbru - Thanks for the review and sorry for the long delay. I went over your comments and believe have addressed the questions/requests. I hope I didn't miss anything, please let me know if I omitted anything.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16116: Rebalance Metrics for AsyncKafkaConsumer [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #15339:
URL: https://github.com/apache/kafka/pull/15339
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org