utafrali commented on code in PR #23471:
URL: https://github.com/apache/kafka/pull/23471#discussion_r4015823943
##########
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java:
##########
@@ -4078,6 +4078,29 @@ public void testTierLagResetsToZeroOnBecomingFollower() {
assertEquals(0,
brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments());
}
+ @Test
+ public void testRemoteDeleteLagResetsToZeroOnBecomingFollower() {
+ remoteLogManager.onLeadershipChange(
+ Set.of(mockPartition(leaderTopicIdPartition)), Set.of(),
topicIds);
+ RemoteLogManager.RLMExpirationTask rlmTask =
+ (RemoteLogManager.RLMExpirationTask)
remoteLogManager.rlmExpirationTask(leaderTopicIdPartition);
+ assertNotNull(rlmTask);
+ rlmTask.updateRemoteDeleteLagWith(2, 1024);
+ assertEquals(1024,
brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteDeleteLagBytes());
+ assertEquals(2,
brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteDeleteLagSegments());
+ // The same node becomes follower now which was the previous leader
+ remoteLogManager.onLeadershipChange(Set.of(),
+ Set.of(mockPartition(leaderTopicIdPartition)), topicIds);
+ assertEquals(0,
brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteDeleteLagBytes());
+ assertEquals(0,
brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteDeleteLagSegments());
+
+ // If the old (now cancelled) expiration task emits the delete-lag
stats, they should be discarded.
+ // Without the isCancelled() guard this re-registers a phantom
non-zero lag that never drains.
+ rlmTask.updateRemoteDeleteLagWith(4, 2048);
+ assertEquals(0,
brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteDeleteLagBytes());
Review Comment:
The test directly invokes `updateRemoteDeleteLagWith` on the cancelled task,
which validates the guard, but it does not cover the actual production
sequence: `removeRemoteTopicPartitionMetrics` is called inside
`onLeadershipChange` before this second `updateRemoteDeleteLagWith` call, so
the gauge that gets re-registered here is a fresh registration rather than a
re-registration after teardown. Consider asserting that the underlying
`KafkaMetric`/gauge is not re-registered in the metrics registry after the
follower transition (e.g. via `brokerTopicStats.topicStats(...)` metric map
inspection), which is what the JIRA describes as the observable symptom.
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -806,6 +806,15 @@ RLMTask rlmCopyTask(TopicIdPartition topicIdPartition) {
return null;
}
+ // VisibleForTesting
+ RLMTask rlmExpirationTask(TopicIdPartition topicIdPartition) {
Review Comment:
Minor: the new `rlmExpirationTask` accessor is only used by the new test.
Since it is symmetric with the existing `rlmCopyTask` helper the addition is
reasonable, but consider tightening the return type to `RLMExpirationTask` so
the test does not need the `(RemoteLogManager.RLMExpirationTask)` cast on line
4082. Same could apply retroactively to `rlmCopyTask`.
##########
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:
The `isCancelled()` check narrows the race window but does not fully close
it: the expiration thread can read `isCancelled() == false`, then
`onLeadershipChange` cancels the task and calls
`removeRemoteTopicPartitionMetrics`, and only then does this method proceed to
`recordRemoteDeleteLagSegments/Bytes`, re-registering a phantom gauge. Since
this mirrors the exact pattern already used for `recordLagStats` on the copy
path, it is fine as-is, but it may be worth a follow-up JIRA (or a one-line
comment here) noting that the guard is best-effort and a fully race-free fix
would require synchronizing cancellation with the metric emission.
--
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]
