LogCleanerManager

LogCleanerManager is created exclusively for LogCleaner.

LogCleanerManager is a KafkaMetricsGroup and registers performance metrics.

Table 1. LogCleanerManager’s Performance Metrics
Metric Name	Description
`max-dirty-percent`
`time-since-last-run-ms`
`uncleanable-bytes`	(for every log directory)
`uncleanable-partitions-count`	(for every log directory)

The performance metrics are registered in kafka.log:type=LogCleanerManager group.

Figure 1. LogCleanerManager in jconsole

LogCleanerManager uses cleaner-offset-checkpoint for the name of the offset checkpoint files for every log directory (in checkpoints registry).

Table 2. LogCleanerManager’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`checkpoints`	`OffsetCheckpointFiles` per log directory Used in updateCheckpoints, maybeTruncateCheckpoint, and alterCheckpointDir to look up the `OffsetCheckpointFile` per log data directory Used in allCleanerCheckpoints to access all `OffsetCheckpointFiles` Used in handleLogDirFailure to exclude a failed log directory from cleaning

LogCleanerManager uses kafka.log.LogCleaner logger.

Tip	Enable `INFO` logging level for `kafka.log.LogCleaner` logger to see what happens inside.

Creating LogCleanerManager Instance

LogCleanerManager takes the following to be created:

Log directories
Logs per Kafka TopicPartition
LogDirFailureChannel

LogCleanerManager initializes the internal registries and counters.

`isCompactAndDelete` Utility

isCompactAndDelete(log: Log): Boolean

isCompactAndDelete flag is enabled (true) when compact and delete cleanup policies are both included in the cleanup.policy configuration property. Otherwise, isCompactAndDelete flag is disabled (false).

Note	`isCompactAndDelete` is used when…FIXME

`cleanableOffsets` Method

cleanableOffsets(
  log: Log,
  topicPartition: TopicPartition,
  lastClean: immutable.Map[TopicPartition, Long],
  now: Long): (Long, Long)

cleanableOffsets…FIXME

Note	`cleanableOffsets` is used when…FIXME

`grabFilthiestCompactedLog` Method

grabFilthiestCompactedLog(time: Time): Option[LogToClean]

grabFilthiestCompactedLog…FIXME

Note	`grabFilthiestCompactedLog` is used when…FIXME

`deletableLogs` Method

deletableLogs(): Iterable[(TopicPartition, Log)]

deletableLogs…FIXME

Note	`deletableLogs` is used when…FIXME

`doneDeleting` Method

doneDeleting(topicPartitions: Iterable[TopicPartition]): Unit

doneDeleting…FIXME

Note	`doneDeleting` is used when…FIXME

`doneCleaning` Method

doneCleaning(
  topicPartition: TopicPartition,
  dataDir: File,
  endOffset: Long): Unit

doneCleaning…FIXME

Note	`doneCleaning` is used exclusively when `CleanerThread` is requested to cleanLog.

`allCleanerCheckpoints` Method

allCleanerCheckpoints: Map[TopicPartition, Long]

allCleanerCheckpoints…FIXME

Note	`allCleanerCheckpoints` is used when: `LogCleanerManager` is requested to grabFilthiestCompactedLog and for the uncleanable-bytes performance metric `LogCleaner` is requested to awaitCleaned (for tests only)

`alterCheckpointDir` Method

alterCheckpointDir(
  topicPartition: TopicPartition,
  sourceLogDir: File,
  destLogDir: File): Unit

alterCheckpointDir…FIXME

Note	`alterCheckpointDir` is used exclusively when `LogCleaner` is requested to alterCheckpointDir.

`handleLogDirFailure` Method

handleLogDirFailure(dir: String): Unit

handleLogDirFailure…FIXME

Note	`handleLogDirFailure` is used exclusively when `LogCleaner` is requested to handleLogDirFailure.

`updateCheckpoints` Method

updateCheckpoints(
  dataDir: File,
  update: Option[(TopicPartition, Long)]): Unit

updateCheckpoints…FIXME

Note	`updateCheckpoints` is used when: `LogCleaner` is requested to updateCheckpoints `LogCleanerManager` is requested to alterCheckpointDir and doneCleaning

`maybeTruncateCheckpoint` Method

maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: Long): Unit

maybeTruncateCheckpoint…FIXME

Note	`maybeTruncateCheckpoint` is used exclusively when `LogCleaner` is requested to maybeTruncateCheckpoint.

LogCleanerManager