chia7712 commented on code in PR #23471:
URL: https://github.com/apache/kafka/pull/23471#discussion_r4042011310
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -1365,11 +1374,20 @@ private void updateMetadataCountAndLogSizeWith(int
metadataCount, long remoteLog
brokerTopicStats.recordRemoteLogSizeBytes(topic, partition,
remoteLogSizeBytes);
}
- private void updateRemoteDeleteLagWith(int segmentsLeftToDelete, long
sizeOfDeletableSegmentsBytes) {
- String topic = topicIdPartition.topic();
- int partition = topicIdPartition.partition();
- brokerTopicStats.recordRemoteDeleteLagSegments(topic, partition,
segmentsLeftToDelete);
- brokerTopicStats.recordRemoteDeleteLagBytes(topic, partition,
sizeOfDeletableSegmentsBytes);
+ // VisibleForTesting
+ void updateRemoteDeleteLagWith(int segmentsLeftToDelete, long
sizeOfDeletableSegmentsBytes) {
+ // Skip emitting metrics for a cancelled task. Otherwise, a task
that is still running in the
+ // expiration thread pool while this replica transitions from
leader to follower can re-register
+ // the delete-lag gauge after onLeadershipChange has already
removed it (see removeRemoteTopicPartitionMetrics),
+ // leaving a phantom non-zero lag that never drains. This mirrors
the guard on the copy path's recordLagStats.
+ // Note: the check is best-effort - it does not synchronize with
cancellation, so it only narrows (does not
+ // fully close) the window where a concurrent leadership change
removes the gauge between this check and the emit.
+ if (!isCancelled()) {
Review Comment:
The other methods also need this guard. I have opened
https://issues.apache.org/jira/browse/KAFKA-21124, and I will file a patch to
address it.
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -1365,11 +1374,18 @@ private void updateMetadataCountAndLogSizeWith(int
metadataCount, long remoteLog
brokerTopicStats.recordRemoteLogSizeBytes(topic, partition,
remoteLogSizeBytes);
}
- private void updateRemoteDeleteLagWith(int segmentsLeftToDelete, long
sizeOfDeletableSegmentsBytes) {
- String topic = topicIdPartition.topic();
- int partition = topicIdPartition.partition();
- brokerTopicStats.recordRemoteDeleteLagSegments(topic, partition,
segmentsLeftToDelete);
- brokerTopicStats.recordRemoteDeleteLagBytes(topic, partition,
sizeOfDeletableSegmentsBytes);
+ // VisibleForTesting
+ void updateRemoteDeleteLagWith(int segmentsLeftToDelete, long
sizeOfDeletableSegmentsBytes) {
+ // Skip emitting metrics for a cancelled task. Otherwise, a task
that is still running in the
+ // expiration thread pool while this replica transitions from
leader to follower can re-register
+ // the delete-lag gauge after onLeadershipChange has already
removed it (see removeRemoteTopicPartitionMetrics),
+ // leaving a phantom non-zero lag that never drains. This mirrors
the guard on the copy path's recordLagStats.
+ if (!isCancelled()) {
+ String topic = topicIdPartition.topic();
+ int partition = topicIdPartition.partition();
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic,
partition, segmentsLeftToDelete);
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, partition,
sizeOfDeletableSegmentsBytes);
+ }
Review Comment:
Yes, this guard is not 100% reliable due to a race condition. I opened
https://issues.apache.org/jira/browse/KAFKA-21125 to tackle it
--
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]
