The new hw is added to incorrect log when ReplicaAlterLogDirsThread is replacing log (fix PlaintextAdminIntegrationTest.testAlterReplicaLogDirs)
Reviewers: Jun Rao
@@ -704,7 +704,11 @@ class Partition(val topicPartition: TopicPartition,
// check if the HW of the partition can now be incremented
// since the replica may already be in the ISR and its LEO has just incremented
val leaderHWIncremented = if (prevFollowerEndOffset != followerReplica.logEndOffset) {
+ // the leader log may be updated by ReplicaAlterLogDirsThread so the following method must be in lock of
+ // leaderIsrUpdateLock to prevent adding new hw to invalid log.
+ inReadLock(leaderIsrUpdateLock) {
leaderLogIfLocal.exists(leaderLog => maybeIncrementLeaderHW(leaderLog, followerFetchTimeMs))
+ }
} else {
false
}
@@ -865,12 +869,11 @@ class Partition(val topicPartition: TopicPartition,
* committed ISR. However, adding additional replicas to the ISR makes it more restrictive and therefor safe. We call
* this set the "maximal" ISR. See KIP-497 for more details
*
- * Returns true if the HW was incremented, and false otherwise.
- * Note There is no need to acquire the leaderIsrUpdate lock here
- * since all callers of this private API acquire that lock
+ * Note There is no need to acquire the leaderIsrUpdate lock here since all callers of this private API acquire that lock
+ *
+ * @return true if the HW was incremented, and false otherwise.
*/
private def maybeIncrementLeaderHW(leaderLog: Log, curTime: Long = time.milliseconds): Boolean = {
- inReadLock(leaderIsrUpdateLock) {
// maybeIncrementLeaderHW is in the hot path, the following code is written to
// avoid unnecessary collection generation
var newHighWatermark = leaderLog.logEndOffsetMetadata
@@ -901,7 +904,6 @@ class Partition(val topicPartition: TopicPartition,
false
}
}
- }
/**
* The low watermark offset value, calculated only if the local replica is the partition leader
