You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/05/15 09:27:56 UTC
[GitHub] [kafka] dajac opened a new pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
dajac opened a new pull request #8672:
URL: https://github.com/apache/kafka/pull/8672
* Update checkpoint files once for all deleted partitions instead of updating them for each deleted partitions.
* Refactor the checkpointing methods to not compute the `logsByDir` all the time. It is now reused as much as possible.
* Refactor the exception handling. Some checkpointing methods were handling `IOException` but the underlying write process already catches them and throws KafkaStorageException instead.
* Reduce the logging in the log cleaner manager. It does not log anymore when a partition is deleted as it is not a useful information.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r446192909
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -1018,6 +1097,15 @@ class LogManager(logDirs: Seq[File],
byDir
}
+ private def logsByDir(dir: File): Map[TopicPartition, Log] = {
Review comment:
Good point.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -1018,6 +1097,15 @@ class LogManager(logDirs: Seq[File],
byDir
}
+ private def logsByDir(dir: File): Map[TopicPartition, Log] = {
+ logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+ }
+
+ private def logsByDir(cachedLogsByDir: Map[String, Map[TopicPartition, Log]],
Review comment:
Good point.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-657555912
@junrao Thanks for your comments. I just pushed an update which addresses your comments.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-646173735
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r453642510
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -2058,13 +2070,37 @@ class ReplicaManagerTest {
val partition = replicaManager.createPartition(tp0)
partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints)
+ val logDirFailureChannel = new LogDirFailureChannel(replicaManager.config.logDirs.size)
+ val logDir = partition.log.get.parentDirFile
+
+ def readRecoveryPointCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.RecoveryPointCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
+ def readLogStartOffsetCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.LogStartOffsetCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
Seq(leaderAndIsrPartitionState(tp0, 1, 0, Seq(0, 1), true)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava
).build()
replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) => ())
+ val batch = TestUtils.records(records = List(
+ new SimpleRecord(10, "k1".getBytes, "v1".getBytes),
+ new SimpleRecord(11, "k2".getBytes, "v2".getBytes)))
+ partition.appendRecordsToLeader(batch, AppendOrigin.Client, requiredAcks = 0)
+ partition.log.get.updateHighWatermark(2L)
+ partition.log.get.maybeIncrementLogStartOffset(1L, LeaderOffsetIncremented)
+ replicaManager.logManager.checkpointLogRecoveryOffsets()
+ replicaManager.logManager.checkpointLogStartOffsets()
+ assertTrue(readRecoveryPointCheckpoint().contains(tp0))
+ assertTrue(readLogStartOffsetCheckpoint().contains(tp0))
Review comment:
Sure.
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -2058,13 +2070,37 @@ class ReplicaManagerTest {
val partition = replicaManager.createPartition(tp0)
partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints)
+ val logDirFailureChannel = new LogDirFailureChannel(replicaManager.config.logDirs.size)
+ val logDir = partition.log.get.parentDirFile
+
+ def readRecoveryPointCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.RecoveryPointCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
+ def readLogStartOffsetCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.LogStartOffsetCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
Seq(leaderAndIsrPartitionState(tp0, 1, 0, Seq(0, 1), true)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava
).build()
replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) => ())
+ val batch = TestUtils.records(records = List(
+ new SimpleRecord(10, "k1".getBytes, "v1".getBytes),
+ new SimpleRecord(11, "k2".getBytes, "v2".getBytes)))
+ partition.appendRecordsToLeader(batch, AppendOrigin.Client, requiredAcks = 0)
+ partition.log.get.updateHighWatermark(2L)
+ partition.log.get.maybeIncrementLogStartOffset(1L, LeaderOffsetIncremented)
+ replicaManager.logManager.checkpointLogRecoveryOffsets()
+ replicaManager.logManager.checkpointLogStartOffsets()
+ assertTrue(readRecoveryPointCheckpoint().contains(tp0))
Review comment:
Sure.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r436791602
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -878,8 +936,7 @@ class LogManager(logDirs: Seq[File],
// Now that replica in source log directory has been successfully renamed for deletion.
// Close the log, update checkpoint files, and enqueue this log to be deleted.
sourceLog.close()
- checkpointRecoveryOffsetsAndCleanSnapshot(sourceLog.parentDirFile, ArrayBuffer.empty)
- checkpointLogStartOffsetsInDir(sourceLog.parentDirFile)
+ checkpointRecoveryAndLogStartOffsetsInDir(sourceLog.parentDirFile)
Review comment:
Well, they are tightly coupled right? The method name of `deleteSnapshotsAfterRecoveryPointCheckpoint` makes it clear that this should be called after the recovery point is checkpointed. Generally, we've had bugs whenever we left it to the callers to make the same multiple calls in sequence in multiple places.
I haven't looked at this PR in detail, so there are probably good reasons to change it. Also keep in mind https://github.com/apache/kafka/pull/7929/files that tries to improve the approach on how we handle this more generally.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -878,8 +936,7 @@ class LogManager(logDirs: Seq[File],
// Now that replica in source log directory has been successfully renamed for deletion.
// Close the log, update checkpoint files, and enqueue this log to be deleted.
sourceLog.close()
- checkpointRecoveryOffsetsAndCleanSnapshot(sourceLog.parentDirFile, ArrayBuffer.empty)
- checkpointLogStartOffsetsInDir(sourceLog.parentDirFile)
+ checkpointRecoveryAndLogStartOffsetsInDir(sourceLog.parentDirFile)
Review comment:
Well, they are tightly coupled right? The method name of `deleteSnapshotsAfterRecoveryPointCheckpoint` makes it clear that this should be called after the recovery point is checkpointed. Generally, we've had bugs whenever we left it to the callers to make the same multiple calls in sequence in multiple places.
I haven't looked at this PR in detail, so there are probably good reasons to change it. Also keep in mind https://github.com/apache/kafka/pull/7929/files which tries to improve the approach on how we handle this more generally.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -878,8 +936,7 @@ class LogManager(logDirs: Seq[File],
// Now that replica in source log directory has been successfully renamed for deletion.
// Close the log, update checkpoint files, and enqueue this log to be deleted.
sourceLog.close()
- checkpointRecoveryOffsetsAndCleanSnapshot(sourceLog.parentDirFile, ArrayBuffer.empty)
- checkpointLogStartOffsetsInDir(sourceLog.parentDirFile)
+ checkpointRecoveryAndLogStartOffsetsInDir(sourceLog.parentDirFile)
Review comment:
Well, they are tightly coupled right? The method name of `deleteSnapshotsAfterRecoveryPointCheckpoint` makes it clear that this should be called after the recovery point is checkpointed. Generally, we've had bugs whenever we left it to the callers to make the same multiple calls in sequence in multiple places.
I haven't looked at this PR in detail, so there are probably good reasons to change it. Also keep in mind #7929 which tries to improve the approach on how we handle this more generally.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-646271944
@hachikuji I just rebased and fixed the build issue. Could you re-trigger jenkins please?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-639770349
ok to test
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r446190937
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -459,7 +459,12 @@ class Partition(val topicPartition: TopicPartition,
}
}
- def delete(): Unit = {
+ /**
+ * Delete the partition. The underlying logs are deleted by default but one can choose to not
+ * delete them automatically and to delete them manually later one. For instance, we do this
+ * in the handling of the StopReplicaRequest to batch the deletions and checkpoint only once.
+ */
+ def delete(deleteLogs: Boolean = true): Unit = {
Review comment:
Yes, we can.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r446192217
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -433,29 +425,34 @@ class ReplicaManager(val config: KafkaConfig,
case HostedPartition.None =>
// Delete log and corresponding folders in case replica manager doesn't hold them anymore.
// This could happen when topic is being deleted while broker is down and recovers.
- stoppedPartitions += topicPartition -> partitionState
+ stoppedPartitions += topicPartition
+ if (deletePartition)
+ deletedPartitions += topicPartition
+ responseMap.put(topicPartition, Errors.NONE)
}
}
// First stop fetchers for all partitions, then stop the corresponding replicas
- val partitions = stoppedPartitions.keySet
- replicaFetcherManager.removeFetcherForPartitions(partitions)
- replicaAlterLogDirsManager.removeFetcherForPartitions(partitions)
+ replicaFetcherManager.removeFetcherForPartitions(stoppedPartitions)
+ replicaAlterLogDirsManager.removeFetcherForPartitions(stoppedPartitions)
- stoppedPartitions.foreach { case (topicPartition, partitionState) =>
- val deletePartition = partitionState.deletePartition
- try {
- stopReplica(topicPartition, deletePartition)
- responseMap.put(topicPartition, Errors.NONE)
- } catch {
+ // Delete the logs and checkpoint
+ logManager.asyncDelete(deletedPartitions, (topicPartition, exception) => {
+ exception match {
case e: KafkaStorageException =>
- stateChangeLogger.error(s"Ignoring StopReplica request (delete=$deletePartition) from " +
+ stateChangeLogger.error(s"Ignoring StopReplica request (delete=true) from " +
s"controller $controllerId with correlation id $correlationId " +
s"epoch $controllerEpoch for partition $topicPartition as the local replica for the " +
- "partition is in an offline log directory", e)
+ "partition is in an offline log directory")
responseMap.put(topicPartition, Errors.KAFKA_STORAGE_ERROR)
+
+ case e =>
+ stateChangeLogger.error(s"Ignoring StopReplica request (delete=true) from " +
+ s"controller $controllerId with correlation id $correlationId " +
+ s"epoch $controllerEpoch for partition $topicPartition due to ${e.getMessage}")
Review comment:
That makes sense.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r445675472
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -459,7 +459,12 @@ class Partition(val topicPartition: TopicPartition,
}
}
- def delete(): Unit = {
+ /**
+ * Delete the partition. The underlying logs are deleted by default but one can choose to not
+ * delete them automatically and to delete them manually later one. For instance, we do this
+ * in the handling of the StopReplicaRequest to batch the deletions and checkpoint only once.
+ */
+ def delete(deleteLogs: Boolean = true): Unit = {
Review comment:
The caller seems to always set deleteLogs to false. Could we just remove this param and the associated code?
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -433,29 +425,34 @@ class ReplicaManager(val config: KafkaConfig,
case HostedPartition.None =>
// Delete log and corresponding folders in case replica manager doesn't hold them anymore.
// This could happen when topic is being deleted while broker is down and recovers.
- stoppedPartitions += topicPartition -> partitionState
+ stoppedPartitions += topicPartition
+ if (deletePartition)
+ deletedPartitions += topicPartition
+ responseMap.put(topicPartition, Errors.NONE)
}
}
// First stop fetchers for all partitions, then stop the corresponding replicas
- val partitions = stoppedPartitions.keySet
- replicaFetcherManager.removeFetcherForPartitions(partitions)
- replicaAlterLogDirsManager.removeFetcherForPartitions(partitions)
+ replicaFetcherManager.removeFetcherForPartitions(stoppedPartitions)
+ replicaAlterLogDirsManager.removeFetcherForPartitions(stoppedPartitions)
- stoppedPartitions.foreach { case (topicPartition, partitionState) =>
- val deletePartition = partitionState.deletePartition
- try {
- stopReplica(topicPartition, deletePartition)
- responseMap.put(topicPartition, Errors.NONE)
- } catch {
+ // Delete the logs and checkpoint
+ logManager.asyncDelete(deletedPartitions, (topicPartition, exception) => {
+ exception match {
case e: KafkaStorageException =>
- stateChangeLogger.error(s"Ignoring StopReplica request (delete=$deletePartition) from " +
+ stateChangeLogger.error(s"Ignoring StopReplica request (delete=true) from " +
s"controller $controllerId with correlation id $correlationId " +
s"epoch $controllerEpoch for partition $topicPartition as the local replica for the " +
- "partition is in an offline log directory", e)
+ "partition is in an offline log directory")
responseMap.put(topicPartition, Errors.KAFKA_STORAGE_ERROR)
+
+ case e =>
+ stateChangeLogger.error(s"Ignoring StopReplica request (delete=true) from " +
+ s"controller $controllerId with correlation id $correlationId " +
+ s"epoch $controllerEpoch for partition $topicPartition due to ${e.getMessage}")
Review comment:
It's probably useful to log the class of the exception too.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -1018,6 +1097,15 @@ class LogManager(logDirs: Seq[File],
byDir
}
+ private def logsByDir(dir: File): Map[TopicPartition, Log] = {
+ logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+ }
+
+ private def logsByDir(cachedLogsByDir: Map[String, Map[TopicPartition, Log]],
Review comment:
This probably should be called logsInDir() ?
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -1018,6 +1097,15 @@ class LogManager(logDirs: Seq[File],
byDir
}
+ private def logsByDir(dir: File): Map[TopicPartition, Log] = {
Review comment:
This probably should be called logsInDir() ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r436519796
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -580,53 +586,105 @@ class LogManager(logDirs: Seq[File],
* to avoid exposing data that have been deleted by DeleteRecordsRequest
*/
def checkpointLogStartOffsets(): Unit = {
- liveLogDirs.foreach(checkpointLogStartOffsetsInDir)
+ val logsByDirCached = logsByDir
+ liveLogDirs.foreach { logDir =>
+ checkpointLogsStartOffsetsInDir(
+ logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty), logDir)
+ }
}
/**
- * Write the recovery checkpoint file for all logs in provided directory and clean older snapshots for provided logs.
- *
- * @param dir the directory in which logs are checkpointed
- * @param logsToCleanSnapshot logs whose snapshots need to be cleaned
- */
+ * Write the checkpoint files for all the provided directories. This is used to cleanup
+ * checkpoints after having deleted partitions.
+ */
+ def checkpoint(logDirs: Set[File]): Unit = {
+ val logsByDirCached = logsByDir
+ logDirs.foreach { logDir =>
+ val partitionToLog = logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty)
+ if (cleaner != null) {
+ cleaner.updateCheckpoints(logDir)
+ }
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, logDir)
+ checkpointLogsStartOffsetsInDir(partitionToLog, logDir)
+ }
+ }
+
+ /**
+ * Clean snapshots of the provided logs in the provided directory.
+ *
+ * @param logsToCleanSnapshot the logs whose snapshots will be cleaned
+ * @param dir the directory in which the logs are
+ */
// Only for testing
- private[log] def checkpointRecoveryOffsetsAndCleanSnapshot(dir: File, logsToCleanSnapshot: Seq[Log]): Unit = {
+ private[log] def cleanSnapshotsInDir(logsToCleanSnapshot: Seq[Log], dir: File): Unit = {
try {
- checkpointLogRecoveryOffsetsInDir(dir)
logsToCleanSnapshot.foreach(_.deleteSnapshotsAfterRecoveryPointCheckpoint())
} catch {
case e: IOException =>
- logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath, s"Disk error while writing to recovery point " +
- s"file in directory $dir", e)
+ logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath,
+ s"Disk error while writing to recovery point file in directory $dir", e)
}
}
- private def checkpointLogRecoveryOffsetsInDir(dir: File): Unit = {
- for {
- partitionToLog <- logsByDir.get(dir.getAbsolutePath)
- checkpoint <- recoveryPointCheckpoints.get(dir)
- } {
- checkpoint.write(partitionToLog.map { case (tp, log) => tp -> log.recoveryPoint })
+ /**
+ * Checkpoint log recovery offsets for all the logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ // Only for testing
+ private[log] def checkpointRecoveryOffsetsInDir(dir: File): Unit = {
+ val partitionToLog = logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, dir)
+ }
+
+ /**
+ * Checkpoint log recovery and start offsets for all logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ private def checkpointRecoveryAndLogStartOffsetsInDir(dir: File): Unit = {
Review comment:
It is a small optimization to avoid computing the `logsByDir` mapping twice when both we checkpoint both at the same time. Computing the mapping is quite expensive when there are many logs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-641249326
@hachikuji I have made another pass on the PR based on your input. Could you please have a second look at it? I am especially interested by getting your view on the overall approach for the checkpointing and the batch deletion.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-650191087
@junrao Thanks for your feedback. I just pushed an update which addresses your comments.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao merged pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
junrao merged pull request #8672:
URL: https://github.com/apache/kafka/pull/8672
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r426083581
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -324,8 +324,17 @@ class ReplicaManager(val config: KafkaConfig,
brokerTopicStats.removeMetrics(topic)
}
- def stopReplica(topicPartition: TopicPartition, deletePartition: Boolean): Unit = {
+ def stopReplica(topicPartition: TopicPartition,
+ deletePartition: Boolean,
+ logDirs: mutable.Set[File]): Unit = {
if (deletePartition) {
+ val log = logManager.getLog(topicPartition)
+ val futureLog = logManager.getLog(topicPartition, isFuture = true)
+
+ // Collect log dirs ofr the partition for future checkpointing
Review comment:
Nit: typo `ofr`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r436704800
##########
File path: core/src/main/scala/kafka/log/LogCleanerManager.scala
##########
@@ -273,7 +272,7 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
* 6. If the partition is already paused, a new call to this function
* will increase the paused count by one.
*/
- def abortAndPauseCleaning(topicPartition: TopicPartition): Unit = {
+ def abortAndPauseCleaning(topicPartition: TopicPartition, partitionDeleted: Boolean = false): Unit = {
Review comment:
Actually, we already have `abortCleaning` for this purpose and `abortCleaning` calls `abortAndPauseCleaning`. I was trying to avoid logging a message when the partition is deleted because it does not bring much and literally flood the log when many partitions are deleted.
While re-looking at this, I have done it differently now. I have found that logs were spread between the LogCleanerManager and the LogManager and that we were logging when resuming cleaning but not all the time. I have consolidated all the cases where we want to log explicitly in helper methods. It also helps with not always having to check if `cleaner != null`. Let me know what you think.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -912,12 +969,9 @@ class LogManager(logDirs: Seq[File],
if (removedLog != null) {
//We need to wait until there is no more cleaning task on the log to be deleted before actually deleting it.
if (cleaner != null && !isFuture) {
- cleaner.abortCleaning(topicPartition)
- cleaner.updateCheckpoints(removedLog.parentDirFile)
+ cleaner.abortCleaning(topicPartition, partitionDeleted = true)
}
removedLog.renameDir(Log.logDeleteDirName(topicPartition))
- checkpointRecoveryOffsetsAndCleanSnapshot(removedLog.parentDirFile, ArrayBuffer.empty)
Review comment:
I totally agree with you. I have tried different ways trying to keep a better encapsulation but I have found a really satisfying way to get there. As you pointed out, the involvement of `Partition` makes this complex.
Here is my best idea so far:
* We remove the calls to `asyncDelete` in the `Partition.delete` method. It seems safe to "delete the partition" while keeping the files on disk while holding the `replicaStateChangeLock` in the `ReplicaManager`.
* We use a `asyncDelete` that takes a batch of logs, deletes them and checkpoint.
* We rename `Partition.delete` to something like `Partition.close` as the log is not really deleted any more.
What do you think?
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -912,12 +969,9 @@ class LogManager(logDirs: Seq[File],
if (removedLog != null) {
//We need to wait until there is no more cleaning task on the log to be deleted before actually deleting it.
if (cleaner != null && !isFuture) {
- cleaner.abortCleaning(topicPartition)
- cleaner.updateCheckpoints(removedLog.parentDirFile)
+ cleaner.abortCleaning(topicPartition, partitionDeleted = true)
}
removedLog.renameDir(Log.logDeleteDirName(topicPartition))
- checkpointRecoveryOffsetsAndCleanSnapshot(removedLog.parentDirFile, ArrayBuffer.empty)
Review comment:
I have implemented it to see. You can check out the last commit.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -878,8 +936,7 @@ class LogManager(logDirs: Seq[File],
// Now that replica in source log directory has been successfully renamed for deletion.
// Close the log, update checkpoint files, and enqueue this log to be deleted.
sourceLog.close()
- checkpointRecoveryOffsetsAndCleanSnapshot(sourceLog.parentDirFile, ArrayBuffer.empty)
- checkpointLogStartOffsetsInDir(sourceLog.parentDirFile)
+ checkpointRecoveryAndLogStartOffsetsInDir(sourceLog.parentDirFile)
Review comment:
Indeed, they are coupled from that perspective. We can keep them together in one method though.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -869,17 +903,18 @@ class LogManager(logDirs: Seq[File],
currentLogs.put(topicPartition, destLog)
if (cleaner != null) {
cleaner.alterCheckpointDir(topicPartition, sourceLog.parentDirFile, destLog.parentDirFile)
- cleaner.resumeCleaning(Seq(topicPartition))
- info(s"Compaction for partition $topicPartition is resumed")
+ resumeCleaning(topicPartition)
}
try {
sourceLog.renameDir(Log.logDeleteDirName(topicPartition))
// Now that replica in source log directory has been successfully renamed for deletion.
// Close the log, update checkpoint files, and enqueue this log to be deleted.
sourceLog.close()
- checkpointRecoveryOffsetsAndCleanSnapshot(sourceLog.parentDirFile, ArrayBuffer.empty)
- checkpointLogStartOffsetsInDir(sourceLog.parentDirFile)
+ val logDir = sourceLog.parentDirFile
+ val logsToCheckpoint = logsByDir(logDir)
+ checkpointRecoveryOffsetsAndCleanSnapshotsInDir(logDir, logsToCheckpoint, ArrayBuffer.empty)
+ checkpointLogStartOffsetsInDir(logDir, logsToCheckpoint)
Review comment:
FYI: This block of code appears twice so it would merit its own method. I haven't done it because that I felt that it would mix too many things in one method and the method name would be too long. Therefore, I left the optimisation of computing and reusing the `logsByDir` on the caller side. I don't feel too strongly about this though.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
junrao commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-657627713
ok to test
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r436523916
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -878,8 +936,7 @@ class LogManager(logDirs: Seq[File],
// Now that replica in source log directory has been successfully renamed for deletion.
// Close the log, update checkpoint files, and enqueue this log to be deleted.
sourceLog.close()
- checkpointRecoveryOffsetsAndCleanSnapshot(sourceLog.parentDirFile, ArrayBuffer.empty)
- checkpointLogStartOffsetsInDir(sourceLog.parentDirFile)
+ checkpointRecoveryAndLogStartOffsetsInDir(sourceLog.parentDirFile)
Review comment:
Actually, we were not cleaning snapshots here. `checkpointRecoveryOffsetsAndCleanSnapshot` was cleaning the snapshots of the logs passed in the second argument. `ArrayBuffer.empty` in our particular case.
Indeed, I thought that there is little sense to couple checkpointing and cleaning the snapshots given that they operate on separate arguments. We were checkpointing based on the first argument and cleaning based on the second one. It is more misleading than anything else.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-629192403
cc @junrao Could you take a look at this one please?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r436115831
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -465,12 +465,15 @@ class LogManager(logDirs: Seq[File],
for ((dir, dirJobs) <- jobs) {
dirJobs.foreach(_.get)
+ val logsInDir = localLogsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+
// update the last flush point
debug(s"Updating recovery points at $dir")
- checkpointRecoveryOffsetsAndCleanSnapshot(dir, localLogsByDir.getOrElse(dir.toString, Map()).values.toSeq)
+ checkpointLogsRecoveryOffsetsInDir(logsInDir, dir)
Review comment:
nit: remove `Logs` from name?
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -580,53 +586,105 @@ class LogManager(logDirs: Seq[File],
* to avoid exposing data that have been deleted by DeleteRecordsRequest
*/
def checkpointLogStartOffsets(): Unit = {
- liveLogDirs.foreach(checkpointLogStartOffsetsInDir)
+ val logsByDirCached = logsByDir
+ liveLogDirs.foreach { logDir =>
+ checkpointLogsStartOffsetsInDir(
+ logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty), logDir)
+ }
}
/**
- * Write the recovery checkpoint file for all logs in provided directory and clean older snapshots for provided logs.
- *
- * @param dir the directory in which logs are checkpointed
- * @param logsToCleanSnapshot logs whose snapshots need to be cleaned
- */
+ * Write the checkpoint files for all the provided directories. This is used to cleanup
+ * checkpoints after having deleted partitions.
+ */
+ def checkpoint(logDirs: Set[File]): Unit = {
+ val logsByDirCached = logsByDir
+ logDirs.foreach { logDir =>
+ val partitionToLog = logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty)
+ if (cleaner != null) {
+ cleaner.updateCheckpoints(logDir)
+ }
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, logDir)
+ checkpointLogsStartOffsetsInDir(partitionToLog, logDir)
+ }
+ }
+
+ /**
+ * Clean snapshots of the provided logs in the provided directory.
+ *
+ * @param logsToCleanSnapshot the logs whose snapshots will be cleaned
+ * @param dir the directory in which the logs are
+ */
// Only for testing
- private[log] def checkpointRecoveryOffsetsAndCleanSnapshot(dir: File, logsToCleanSnapshot: Seq[Log]): Unit = {
+ private[log] def cleanSnapshotsInDir(logsToCleanSnapshot: Seq[Log], dir: File): Unit = {
try {
- checkpointLogRecoveryOffsetsInDir(dir)
logsToCleanSnapshot.foreach(_.deleteSnapshotsAfterRecoveryPointCheckpoint())
} catch {
case e: IOException =>
- logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath, s"Disk error while writing to recovery point " +
- s"file in directory $dir", e)
+ logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath,
+ s"Disk error while writing to recovery point file in directory $dir", e)
}
}
- private def checkpointLogRecoveryOffsetsInDir(dir: File): Unit = {
- for {
- partitionToLog <- logsByDir.get(dir.getAbsolutePath)
- checkpoint <- recoveryPointCheckpoints.get(dir)
- } {
- checkpoint.write(partitionToLog.map { case (tp, log) => tp -> log.recoveryPoint })
+ /**
+ * Checkpoint log recovery offsets for all the logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ // Only for testing
+ private[log] def checkpointRecoveryOffsetsInDir(dir: File): Unit = {
+ val partitionToLog = logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, dir)
+ }
+
+ /**
+ * Checkpoint log recovery and start offsets for all logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ private def checkpointRecoveryAndLogStartOffsetsInDir(dir: File): Unit = {
+ val partitionToLog = logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, dir)
+ checkpointLogsStartOffsetsInDir(partitionToLog, dir)
+ }
+
+ /**
+ * Checkpoint log recovery offsets for all the provided logs in the provided directory.
+ *
+ * @param logs the logs and logs to be checkpointed
+ * @param dir the directory in which logs are checkpointed
+ */
+ private def checkpointLogsRecoveryOffsetsInDir(logs: Map[TopicPartition, Log],
+ dir: File): Unit = {
+ try {
+ recoveryPointCheckpoints.get(dir).foreach { checkpoint =>
+ val recoveryOffsets = logs.map { case (tp, log) => tp -> log.recoveryPoint }
+ checkpoint.write(recoveryOffsets)
+ }
+ } catch {
+ case e: KafkaStorageException =>
+ error(s"Disk error while writing recovery offsets checkpoint in directory $dir: ${e.getMessage}")
}
}
/**
- * Checkpoint log start offset for all logs in provided directory.
+ * Checkpoint log start offsets for all the provided logs in the provided directory.
+ *
+ * @param logs the partitions and logs to be checkpointed
+ * @param dir the directory in which logs are checkpointed
*/
- private def checkpointLogStartOffsetsInDir(dir: File): Unit = {
- for {
- partitionToLog <- logsByDir.get(dir.getAbsolutePath)
- checkpoint <- logStartOffsetCheckpoints.get(dir)
- } {
- try {
- val logStartOffsets = partitionToLog.collect {
+ private def checkpointLogsStartOffsetsInDir(logs: Map[TopicPartition, Log],
+ dir: File): Unit = {
+ try {
+ logStartOffsetCheckpoints.get(dir).foreach { checkpoint =>
+ val logStartOffsets = logs.collect {
case (k, log) if log.logStartOffset > log.logSegments.head.baseOffset => k -> log.logStartOffset
Review comment:
nit: k -> tp?
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -580,53 +586,105 @@ class LogManager(logDirs: Seq[File],
* to avoid exposing data that have been deleted by DeleteRecordsRequest
*/
def checkpointLogStartOffsets(): Unit = {
- liveLogDirs.foreach(checkpointLogStartOffsetsInDir)
+ val logsByDirCached = logsByDir
+ liveLogDirs.foreach { logDir =>
+ checkpointLogsStartOffsetsInDir(
+ logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty), logDir)
+ }
}
/**
- * Write the recovery checkpoint file for all logs in provided directory and clean older snapshots for provided logs.
- *
- * @param dir the directory in which logs are checkpointed
- * @param logsToCleanSnapshot logs whose snapshots need to be cleaned
- */
+ * Write the checkpoint files for all the provided directories. This is used to cleanup
+ * checkpoints after having deleted partitions.
+ */
+ def checkpoint(logDirs: Set[File]): Unit = {
+ val logsByDirCached = logsByDir
+ logDirs.foreach { logDir =>
+ val partitionToLog = logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty)
+ if (cleaner != null) {
+ cleaner.updateCheckpoints(logDir)
+ }
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, logDir)
+ checkpointLogsStartOffsetsInDir(partitionToLog, logDir)
+ }
+ }
+
+ /**
+ * Clean snapshots of the provided logs in the provided directory.
+ *
+ * @param logsToCleanSnapshot the logs whose snapshots will be cleaned
+ * @param dir the directory in which the logs are
+ */
// Only for testing
- private[log] def checkpointRecoveryOffsetsAndCleanSnapshot(dir: File, logsToCleanSnapshot: Seq[Log]): Unit = {
+ private[log] def cleanSnapshotsInDir(logsToCleanSnapshot: Seq[Log], dir: File): Unit = {
try {
- checkpointLogRecoveryOffsetsInDir(dir)
logsToCleanSnapshot.foreach(_.deleteSnapshotsAfterRecoveryPointCheckpoint())
} catch {
case e: IOException =>
- logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath, s"Disk error while writing to recovery point " +
- s"file in directory $dir", e)
+ logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath,
+ s"Disk error while writing to recovery point file in directory $dir", e)
}
}
- private def checkpointLogRecoveryOffsetsInDir(dir: File): Unit = {
- for {
- partitionToLog <- logsByDir.get(dir.getAbsolutePath)
- checkpoint <- recoveryPointCheckpoints.get(dir)
- } {
- checkpoint.write(partitionToLog.map { case (tp, log) => tp -> log.recoveryPoint })
+ /**
+ * Checkpoint log recovery offsets for all the logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ // Only for testing
+ private[log] def checkpointRecoveryOffsetsInDir(dir: File): Unit = {
+ val partitionToLog = logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, dir)
+ }
+
+ /**
+ * Checkpoint log recovery and start offsets for all logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ private def checkpointRecoveryAndLogStartOffsetsInDir(dir: File): Unit = {
+ val partitionToLog = logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
Review comment:
There are a couple place where we have this code. Maybe it make sense to have an overload which accepts the directory.
By the way, `logsByDir` can be private.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -878,8 +936,7 @@ class LogManager(logDirs: Seq[File],
// Now that replica in source log directory has been successfully renamed for deletion.
// Close the log, update checkpoint files, and enqueue this log to be deleted.
sourceLog.close()
- checkpointRecoveryOffsetsAndCleanSnapshot(sourceLog.parentDirFile, ArrayBuffer.empty)
- checkpointLogStartOffsetsInDir(sourceLog.parentDirFile)
+ checkpointRecoveryAndLogStartOffsetsInDir(sourceLog.parentDirFile)
Review comment:
Previously we were cleaning snapshots here. I agree it was probably unnecessary, but there seems little harm in it. I guess the main thing is that logically it makes little sense to couple these operations together even if the events which trigger them are often the same. Is that what you were thinking?
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -324,8 +324,17 @@ class ReplicaManager(val config: KafkaConfig,
brokerTopicStats.removeMetrics(topic)
}
- def stopReplica(topicPartition: TopicPartition, deletePartition: Boolean): Unit = {
+ def stopReplica(topicPartition: TopicPartition,
Review comment:
I believe it would be straightforward to update the uses of this method in the test case to use `stopReplicas` instead. Then we could make this private, which would make the side effect less annoying.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -912,12 +969,9 @@ class LogManager(logDirs: Seq[File],
if (removedLog != null) {
//We need to wait until there is no more cleaning task on the log to be deleted before actually deleting it.
if (cleaner != null && !isFuture) {
- cleaner.abortCleaning(topicPartition)
- cleaner.updateCheckpoints(removedLog.parentDirFile)
+ cleaner.abortCleaning(topicPartition, partitionDeleted = true)
}
removedLog.renameDir(Log.logDeleteDirName(topicPartition))
- checkpointRecoveryOffsetsAndCleanSnapshot(removedLog.parentDirFile, ArrayBuffer.empty)
Review comment:
With this change, it is up to the caller to ensure offsets are checkpointed correctly after a deletion. It would be better from an encapsulation perspective to keep `LogManager` in charge of that. One idea would be to offer a batched `asyncDelete`. The involvement of `Partition` may make this a bit complex though.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -580,53 +586,105 @@ class LogManager(logDirs: Seq[File],
* to avoid exposing data that have been deleted by DeleteRecordsRequest
*/
def checkpointLogStartOffsets(): Unit = {
- liveLogDirs.foreach(checkpointLogStartOffsetsInDir)
+ val logsByDirCached = logsByDir
+ liveLogDirs.foreach { logDir =>
+ checkpointLogsStartOffsetsInDir(
+ logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty), logDir)
+ }
}
/**
- * Write the recovery checkpoint file for all logs in provided directory and clean older snapshots for provided logs.
- *
- * @param dir the directory in which logs are checkpointed
- * @param logsToCleanSnapshot logs whose snapshots need to be cleaned
- */
+ * Write the checkpoint files for all the provided directories. This is used to cleanup
+ * checkpoints after having deleted partitions.
+ */
+ def checkpoint(logDirs: Set[File]): Unit = {
+ val logsByDirCached = logsByDir
+ logDirs.foreach { logDir =>
+ val partitionToLog = logsByDirCached.getOrElse(logDir.getAbsolutePath, Map.empty)
+ if (cleaner != null) {
+ cleaner.updateCheckpoints(logDir)
+ }
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, logDir)
+ checkpointLogsStartOffsetsInDir(partitionToLog, logDir)
+ }
+ }
+
+ /**
+ * Clean snapshots of the provided logs in the provided directory.
+ *
+ * @param logsToCleanSnapshot the logs whose snapshots will be cleaned
+ * @param dir the directory in which the logs are
+ */
// Only for testing
- private[log] def checkpointRecoveryOffsetsAndCleanSnapshot(dir: File, logsToCleanSnapshot: Seq[Log]): Unit = {
+ private[log] def cleanSnapshotsInDir(logsToCleanSnapshot: Seq[Log], dir: File): Unit = {
try {
- checkpointLogRecoveryOffsetsInDir(dir)
logsToCleanSnapshot.foreach(_.deleteSnapshotsAfterRecoveryPointCheckpoint())
} catch {
case e: IOException =>
- logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath, s"Disk error while writing to recovery point " +
- s"file in directory $dir", e)
+ logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath,
+ s"Disk error while writing to recovery point file in directory $dir", e)
}
}
- private def checkpointLogRecoveryOffsetsInDir(dir: File): Unit = {
- for {
- partitionToLog <- logsByDir.get(dir.getAbsolutePath)
- checkpoint <- recoveryPointCheckpoints.get(dir)
- } {
- checkpoint.write(partitionToLog.map { case (tp, log) => tp -> log.recoveryPoint })
+ /**
+ * Checkpoint log recovery offsets for all the logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ // Only for testing
+ private[log] def checkpointRecoveryOffsetsInDir(dir: File): Unit = {
+ val partitionToLog = logsByDir.getOrElse(dir.getAbsolutePath, Map.empty)
+ checkpointLogsRecoveryOffsetsInDir(partitionToLog, dir)
+ }
+
+ /**
+ * Checkpoint log recovery and start offsets for all logs in the provided directory.
+ *
+ * @param dir the directory in which logs are checkpointed
+ */
+ private def checkpointRecoveryAndLogStartOffsetsInDir(dir: File): Unit = {
Review comment:
Given that we split `checkpointRecoveryOffsetsAndCleanSnapshot`, I'm curious why we added this.
##########
File path: core/src/main/scala/kafka/log/LogCleanerManager.scala
##########
@@ -273,7 +272,7 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
* 6. If the partition is already paused, a new call to this function
* will increase the paused count by one.
*/
- def abortAndPauseCleaning(topicPartition: TopicPartition): Unit = {
+ def abortAndPauseCleaning(topicPartition: TopicPartition, partitionDeleted: Boolean = false): Unit = {
Review comment:
It is a little weird to abort _and pause_ cleaning if the partition is getting deleted. Maybe we should have a separate method like `removePartition` or something like that.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r436532135
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -324,8 +324,17 @@ class ReplicaManager(val config: KafkaConfig,
brokerTopicStats.removeMetrics(topic)
}
- def stopReplica(topicPartition: TopicPartition, deletePartition: Boolean): Unit = {
+ def stopReplica(topicPartition: TopicPartition,
Review comment:
I agree.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#issuecomment-653514573
@hachikuji or @junrao Could one of you two help me with getting this merged?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #8672: KAFKA-10002; Improve performances of StopReplicaRequest with large number of partitions to be deleted
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #8672:
URL: https://github.com/apache/kafka/pull/8672#discussion_r453132769
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -2058,13 +2070,37 @@ class ReplicaManagerTest {
val partition = replicaManager.createPartition(tp0)
partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints)
+ val logDirFailureChannel = new LogDirFailureChannel(replicaManager.config.logDirs.size)
+ val logDir = partition.log.get.parentDirFile
+
+ def readRecoveryPointCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.RecoveryPointCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
+ def readLogStartOffsetCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.LogStartOffsetCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
Seq(leaderAndIsrPartitionState(tp0, 1, 0, Seq(0, 1), true)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava
).build()
replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) => ())
+ val batch = TestUtils.records(records = List(
+ new SimpleRecord(10, "k1".getBytes, "v1".getBytes),
+ new SimpleRecord(11, "k2".getBytes, "v2".getBytes)))
+ partition.appendRecordsToLeader(batch, AppendOrigin.Client, requiredAcks = 0)
+ partition.log.get.updateHighWatermark(2L)
+ partition.log.get.maybeIncrementLogStartOffset(1L, LeaderOffsetIncremented)
+ replicaManager.logManager.checkpointLogRecoveryOffsets()
+ replicaManager.logManager.checkpointLogStartOffsets()
+ assertTrue(readRecoveryPointCheckpoint().contains(tp0))
+ assertTrue(readLogStartOffsetCheckpoint().contains(tp0))
Review comment:
Could we verify the expected logStartOffset?
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -2058,13 +2070,37 @@ class ReplicaManagerTest {
val partition = replicaManager.createPartition(tp0)
partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints)
+ val logDirFailureChannel = new LogDirFailureChannel(replicaManager.config.logDirs.size)
+ val logDir = partition.log.get.parentDirFile
+
+ def readRecoveryPointCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.RecoveryPointCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
+ def readLogStartOffsetCheckpoint(): Map[TopicPartition, Long] = {
+ new OffsetCheckpointFile(new File(logDir, LogManager.LogStartOffsetCheckpointFile),
+ logDirFailureChannel).read()
+ }
+
val becomeLeaderRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch,
Seq(leaderAndIsrPartitionState(tp0, 1, 0, Seq(0, 1), true)).asJava,
Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava
).build()
replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) => ())
+ val batch = TestUtils.records(records = List(
+ new SimpleRecord(10, "k1".getBytes, "v1".getBytes),
+ new SimpleRecord(11, "k2".getBytes, "v2".getBytes)))
+ partition.appendRecordsToLeader(batch, AppendOrigin.Client, requiredAcks = 0)
+ partition.log.get.updateHighWatermark(2L)
+ partition.log.get.maybeIncrementLogStartOffset(1L, LeaderOffsetIncremented)
+ replicaManager.logManager.checkpointLogRecoveryOffsets()
+ replicaManager.logManager.checkpointLogStartOffsets()
+ assertTrue(readRecoveryPointCheckpoint().contains(tp0))
Review comment:
Could we verify the expected recovery offset?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org