KAFKA-16709: abortAndPauseCleaning only when future log is not existed

[showuon]

When doing alter replica logDirs, we'll create a future log and pause log cleaning for the partition( here). And this log cleaning pausing will resume after alter replica logDirs completes (here). And when in the resuming log cleaning, we'll decrement 1 for the LogCleaningPaused count. Once the count reached 0, the cleaning pause is really resuming. (here). For more explanation about the logCleaningPaused state can check here.

But, there's still one factor that could increase the LogCleaningPaused count: leadership change (here). When there's a leadership change, we'll check if there's a future log in this partition, if so, we'll create future log and pauseCleaning (LogCleaningPaused count + 1). So, if during the alter replica logDirs:

1. alter replica logDirs for tp0 triggered (LogCleaningPaused count = 1)
2. tp0 leadership changed (LogCleaningPaused count = 2)
3. alter replica logDirs completes, resuming logCleaning (LogCleaningPaused count = 1)
4. LogCleaning keeps paused because the count is always >  0

This PR fixes this issue by only abortAndPauseCleaning when future log is not existed. We did the same check in alterReplicaLogDirs. So this change can make sure there's only 1 abortAndPauseCleaning for either abortAndPauseCleaning or maybeAddLogDirFetchers. Tests also added.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[showuon]

@soarez , call for review. Thanks.

[soarez]

Thanks for looking into this @showuon , I've seen similar issues in production.

I'm not sure if logManager.abortAndPauseCleaning() should be called from maybeAddLogDirFetchers, since it AFAICT the fetcher will call it before starting.

[soarez]

I assume this comment refers to setting (populating) partition.futureLog, which is only written to via a call partition.createLogIfNotExists().
Since we're replacing the call to partition.createLogIfNotExists() with partition.maybeCreateFutureReplica(), this comment should be moved there?

[showuon]

We need this as described in this comment: #15951 (comment)

[soarez]

partition.maybeCreateFutureReplica() only calls partition.createLogIfNotExists() if the future replica doesn't yet exist. If that happens, partition.futureLog won't be set, so we need to call partition.setLog(futureLog, true)?

[soarez]

Sincre this section is inside a block of logManager.getLog(topicPartition, isFuture = true).foreach { futureLog =>, doesn't that mean this only runs if the future replica exists?

[showuon]

partition.maybeCreateFutureReplica() only calls partition.createLogIfNotExists() if the future replica doesn't yet exist. If that happens, partition.futureLog won't be set, so we need to call partition.setLog(futureLog, true)?

partition.setLog(futureLog, true) is not necessary because when in partition.createLogIfNotExists(), we'll also set futureLog:

  if (isFutureReplica) {
      this.futureLog = Some(maybeCreate(this.futureLog))

Sincre this section is inside a block of logManager.getLog(topicPartition, isFuture = true).foreach { futureLog =>, doesn't that mean this only runs if the future replica exists?

Yes, normally, when we created future log is in this path:

1. ReplicaManager#alterReplicaLogDirs
2. partition.maybeCreateFutureReplica
3. partition.createLogIfNotExists
4. partition.createLog
5. logManager.getOrCreateLog

So in the end, we'll have future log added in both logManager#futureLogs map and partition#futureLog` map.

But it's possible that the future log only exists in logManager#futureLogs map, but not in partition#futureLog map, when the future log is created, and before alter logDir completed, the broker restarted. So, after restarted, we'll add the future log into logManager map during loadLog. But the partition#futureLog map won't be updated, until we got leadership update and ReplicaManager#maybeAddLogDirFetchers is called. In this case, the partition.maybeCreateFutureReplica will not create the future log since it's existed, but the partition#futureLog will get updated with the one in logManager.

That means, we have to do the partition.createLogIfNotExists() here.

[soarez]

That means, we have to do the partition.createLogIfNotExists() here.

Given that partition.maybeCreateFutureReplica() only calls partition.createLogIfNotExists() if the future replica doesn't yet exist, and that it's possible that the future log only exists in logManager#futureLogs map, but not in partition#futureLog map – do we need to call partition.createLogIfNotExists() directly instead of ``partition.maybeCreateFutureReplica()` ?

[showuon]

We need a way to determine if partition#futureLog map exists or not, so if we call partition.createLogIfNotExists(), we can't know if it is created or not. partition.maybeCreateFutureReplica() returns a boolean value to notify us what we need.

That is, partition.maybeCreateFutureReplica() only calls partition.createLogIfNotExists() when the partition#futureLog map doesn't contain the partition. That's fine because there is no chances that futureLog exists in partition#futureLog but not in logManager#futureLogs map.

Sorry, maybe I didn't get your question here. Could you explain again if I misunderstand it?

[soarez]

That's fine because there is no chances that futureLog exists in partition#futureLog but not in logManager#futureLogs map.

The scenario I was thinking of is when the broker starts up, logManager loads the future log, so futureLog exists in partition#futureLog but not in logManager#futureLogs map yet. Only later the when the broker catches up with metatada (ReplicaManager#applyDelta) or receives a LeaderAndIsr request (becomeLeaderOrFollower) and this method maybeAddLogDirFetchers is called, is when we need to make sure partition#futureLog is populated too.

But you're right, my confusion here was with where maybeCreateFutureReplica checks if the futureLog already exists, it checks in itself (Partition) not in LogManager, so this makes sense.

[chia7712]

Is it possible that LeaderAndIsrRequest carries a different topic id ?

[showuon]

Yes, it's possible, and it's been handled in ReplicaManager here.

[chia7712]

Pardon me, Do you mean that topic id of future log can be different to topic id of log?

[showuon]

No, the future log is just a follower of the original log, so it must have the same topic ID as the original one.

[chia7712]

My point was this test case will produce different topic ids. For example, the following assert will fail

      assertEquals(spyLogManager.getLog(tp0, isFuture = false).get.topicId,
        spyLogManager.getLog(tp0, isFuture = true).get.topicId)

It seems becomeLeaderOrFollower set the topic id of "log", and the mock return a different topic id of future log. IIRC (and @showuon your confirm) they should have same topic ID, and so I'm a bit confused by this test case.

Please feel free to correct me if I misunderstand anything.

[showuon]

Ah, I know what you're talking about now. The reason it won't fail is because the topicId we feed into the partition object is None. So the topicId consistency check will always pass because the original topicId is not set. I've updated the test and also verify the response has no errors in this commit: 4a1b76d . Thanks.

[showuon]

Thanks to the failing test, I found I was wrong. We should always add the partition into fetch thread no matter we created the future log or not since before maybeAddLogDirFetchers is called, the fetchers are all removed.

[showuon]

@chia7712 , I've updated the PR. Please take a look again when available. Thanks.

[soarez]

Sorry, I clicked the wrong window, did not mean to approve, I'm still reviewing this

[soarez]

Thanks for fixing this @showuon.

This PR looks good to me!

I ran all the tests that failed in CI locally and they all pass.

[chia7712]

If maybeCreateFutureReplica return false, we assume another thread already add the future log to partition and invoke alter thread. Hence, we don't need to abort cleaning since another thread does it.

However, adding alter thread (replicaAlterLogDirsManager.addFetcherForPartitions(futureReplicasAndInitialOffset)) is not in this check. Is it possible that alter thread, which is invoked by another thread, just remove the future log and then this thread add the topic partition to replicaAlterLogDirsManager? It seems to me that alter thread will get fail as future log of partition is gone.

[showuon]

However, adding alter thread (replicaAlterLogDirsManager.addFetcherForPartitions(futureReplicasAndInitialOffset)) is not in this check. Is it possible that alter thread, which is invoked by another thread, just remove the future log and then this thread add the topic partition to replicaAlterLogDirsManager? It seems to me that alter thread will get fail as future log of partition is gone.

That's possible. But I think that's fine because the removal of future log could because:

1. alter logDir completes. In this case, the new leaderAndIsr request or topic partition update will updated and this fetcher will be removed then in ReplicaManager#makeLeader or makeFollower.
2. Another log failure happened. In this case the createLogIfInexsted will fail, too.

[chia7712]

oh, my previous comments is incorrect. Both alterReplicaLogDirs and maybeAddLogDirFetchers are in replicaStateChangeLock, so the race condition I described should not happen.

However, I'm thinking whether it is fine to add alter thread by maybeAddLogDirFetchers even though the future log of partition is already created by another thread. Although no new alter thread will be created as BrokerIdAndFetcherId is identical.

In short, alterReplicaLogDirs adds alter thread [0] only if it succeeds to create future log of partition. Maybe maybeAddLogDirFetchers should follow same rule? Or we can add comments to say "that is fine as replicaAlterLogDirsManager.addFetcherForPartitions will be a no-op in this case?

[0]

kafka/core/src/main/scala/kafka/server/ReplicaManager.scala

Line 1198 in 5552f5c

if (partition.maybeCreateFutureReplica(destinationDir, highWatermarkCheckpoints)) {

[showuon]

@chia7712 , thanks for the comment.
For this:

In short, alterReplicaLogDirs adds alter thread [0] only if it succeeds to create future log of partition. Maybe maybeAddLogDirFetchers should follow same rule? Or we can add comments to say "that is fine as replicaAlterLogDirsManager.addFetcherForPartitions will be a no-op in this case?

I chose latter option because if we only create fetcher when future log is inexisted, it might cause potential side effect that this fetcher is removed when leadership change, but not get added later. I've added the comment in this commit: 0d78e49 . Thanks.

[chia7712]

LGTM

[showuon]

Failed tests are unrelated.

[junrao]

@showuon : Thanks for the PR and sorry for the late review. Just a few minor comments.

[junrao]

Could we add the new param to the javadoc?

[junrao]

When the leader changes, we need to propagate the new leader epoch to ReplicaAlterLogDirsThread (see #8223). So, the operation is not a no-op?

[junrao]

Hmm, becomeLeaderOrFollower(), alterReplicaLogDirs(), and applyDelta() are done under the replicaStateChangeLock. Is it really possible for another thread to add fetcher for the future log?

[junrao]

if it's not existed => if it doesn't exist