) () -> recordsIterator) {
+ if (record.hasKey() && record.offset() >= startOffset) {
+ if (map.size() < maxDesiredMapSize) {
+ map.put(record.key(), record.offset());
+ } else {
+ return true;
+ }
+ }
+ stats.indexMessagesRead(1);
+ }
+ }
+ }
+ }
+
+ if (batch.lastOffset() >= startOffset)
+ map.updateLatestOffset(batch.lastOffset());
+ }
+ int bytesRead = records.validBytes();
+ position += bytesRead;
+ stats.indexBytesRead(bytesRead);
+
+ // if we didn't read even one complete message, our read buffer may be too small
+ if (position == startPosition)
+ growBuffersOrFail(segment.log(), position, maxLogMessageSize, records);
+ }
+
+ // In the case of offsets gap, fast forward to latest expected offset in this segment.
+ map.updateLatestOffset(nextSegmentStartOffset - 1L);
+
+ restoreBuffers();
+
+ return false;
+ }
+}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanerConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanerConfig.java
index 4ae7779f5e84a..b3f814628640a 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanerConfig.java
@@ -16,8 +16,10 @@
*/
package org.apache.kafka.storage.internals.log;
+import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.server.config.ServerTopicConfigSynonyms;
import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
@@ -129,6 +131,22 @@ public CleanerConfig(int numThreads,
this.enableCleaner = enableCleaner;
}
+ /**
+ * Create an instance of this class.
+ *
+ * @param config the {@link AbstractConfig} instance for initialize all variables
+ */
+ public CleanerConfig(AbstractConfig config) {
+ this.numThreads = config.getInt(LOG_CLEANER_THREADS_PROP);
+ this.dedupeBufferSize = config.getLong(LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP);
+ this.dedupeBufferLoadFactor = config.getDouble(LOG_CLEANER_DEDUPE_BUFFER_LOAD_FACTOR_PROP);
+ this.ioBufferSize = config.getInt(LOG_CLEANER_IO_BUFFER_SIZE_PROP);
+ this.maxMessageSize = config.getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG);
+ this.maxIoBytesPerSecond = config.getDouble(LOG_CLEANER_IO_MAX_BYTES_PER_SECOND_PROP);
+ this.backoffMs = config.getLong(LOG_CLEANER_BACKOFF_MS_PROP);
+ this.enableCleaner = config.getBoolean(LOG_CLEANER_ENABLE_PROP);
+ }
+
public String hashAlgorithm() {
return HASH_ALGORITHM;
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanerStats.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanerStats.java
new file mode 100644
index 0000000000000..3dff18b930b76
--- /dev/null
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/CleanerStats.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import org.apache.kafka.common.utils.Time;
+
+/**
+ * A simple struct for collecting stats about log cleaning.
+ */
+public class CleanerStats {
+ long bytesRead = 0L;
+ long bytesWritten = 0L;
+ long mapBytesRead = 0L;
+ long messagesRead = 0L;
+ long invalidMessagesRead = 0L;
+ long messagesWritten = 0L;
+ double bufferUtilization = 0.0d;
+
+ private long mapCompleteTime = -1L;
+ private long endTime = -1L;
+ private long mapMessagesRead = 0L;
+
+ private final Time time;
+ private final long startTime;
+
+ public CleanerStats(Time time) {
+ this.time = time;
+ startTime = time.milliseconds();
+ }
+
+ public void readMessages(int messagesRead, int bytesRead) {
+ this.messagesRead += messagesRead;
+ this.bytesRead += bytesRead;
+ }
+
+ public void invalidMessage() {
+ invalidMessagesRead += 1;
+ }
+
+ public void recopyMessages(int messagesWritten, int bytesWritten) {
+ this.messagesWritten += messagesWritten;
+ this.bytesWritten += bytesWritten;
+ }
+
+ public void indexMessagesRead(int size) {
+ mapMessagesRead += size;
+ }
+
+ public void indexBytesRead(int size) {
+ mapBytesRead += size;
+ }
+
+ public void indexDone() {
+ mapCompleteTime = time.milliseconds();
+ }
+
+ public void allDone() {
+ endTime = time.milliseconds();
+ }
+
+ public double elapsedSecs() {
+ return (endTime - startTime) / 1000.0;
+ }
+
+ public double elapsedIndexSecs() {
+ return (mapCompleteTime - startTime) / 1000.0;
+ }
+
+ // Only for testing
+ public long startTime() {
+ return startTime;
+ }
+
+ // Only for testing
+ public long endTime() {
+ return endTime;
+ }
+
+ // Only for testing
+ public void setEndTime(long endTime) {
+ this.endTime = endTime;
+ }
+
+ // Only for testing
+ public long bytesRead() {
+ return bytesRead;
+ }
+
+ // Only for testing
+ public long bytesWritten() {
+ return bytesWritten;
+ }
+
+ // Only for testing
+ public long mapMessagesRead() {
+ return mapMessagesRead;
+ }
+
+ // Only for testing
+ public long messagesRead() {
+ return messagesRead;
+ }
+
+ // Only for testing
+ public long invalidMessagesRead() {
+ return invalidMessagesRead;
+ }
+
+ // Only for testing
+ public long messagesWritten() {
+ return messagesWritten;
+ }
+
+ // Only for testing
+ public double bufferUtilization() {
+ return bufferUtilization;
+ }
+
+ // Only for testing
+ public void setBufferUtilization(double bufferUtilization) {
+ this.bufferUtilization = bufferUtilization;
+ }
+}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleaner.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleaner.java
new file mode 100644
index 0000000000000..2c0086c4700a1
--- /dev/null
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleaner.java
@@ -0,0 +1,650 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.internals.log;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.errors.KafkaStorageException;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.config.BrokerReconfigurable;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.server.metrics.KafkaMetricsGroup;
+import org.apache.kafka.server.util.ShutdownableThread;
+import org.apache.kafka.storage.internals.utils.Throttler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.DigestException;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * The cleaner is responsible for removing obsolete records from logs which have the "compact" retention strategy.
+ * A message with key K and offset O is obsolete if there exists a message with key K and offset O' such that O < O'.
+ *
+ * Each log can be thought of being split into two sections of segments: a "clean" section which has previously been cleaned followed by a
+ * "dirty" section that has not yet been cleaned. The dirty section is further divided into the "cleanable" section followed by an "uncleanable" section.
+ * The uncleanable section is excluded from cleaning. The active log segment is always uncleanable. If there is a
+ * compaction lag time set, segments whose largest message timestamp is within the compaction lag time of the cleaning operation are also uncleanable.
+ *
+ * The cleaning is carried out by a pool of background threads. Each thread chooses the dirtiest log that has the "compact" retention policy
+ * and cleans that. The dirtiness of the log is guessed by taking the ratio of bytes in the dirty section of the log to the total bytes in the log.
+ *
+ * To clean a log the cleaner first builds a mapping of key=>last_offset for the dirty section of the log. See {@link OffsetMap} for details of
+ * the implementation of the mapping.
+ *
+ * Once the key=>last_offset map is built, the log is cleaned by recopying each log segment but omitting any key that appears in the offset map with a
+ * higher offset than what is found in the segment (i.e. messages with a key that appears in the dirty section of the log).
+ *
+ * To avoid segments shrinking to very small sizes with repeated cleanings we implement a rule by which if we will merge successive segments when
+ * doing a cleaning if their log and index size are less than the maximum log and index size prior to the clean beginning.
+ *
+ * Cleaned segments are swapped into the log as they become available.
+ *
+ * One nuance that the cleaner must handle is log truncation. If a log is truncated while it is being cleaned the cleaning of that log is aborted.
+ *
+ * Messages with null payload are treated as deletes for the purpose of log compaction. This means that they receive special treatment by the cleaner.
+ * The cleaner will only retain delete records for a period of time to avoid accumulating space indefinitely. This period of time is configurable on a per-topic
+ * basis and is measured from the time the segment enters the clean portion of the log (at which point any prior message with that key has been removed).
+ * Delete markers in the clean section of the log that are older than this time will not be retained when log segments are being recopied as part of cleaning.
+ * This time is tracked by setting the base timestamp of a record batch with delete markers when the batch is recopied in the first cleaning that encounters
+ * it. The relative timestamps of the records in the batch are also modified when recopied in this cleaning according to the new base timestamp of the batch.
+ *
+ * Note that cleaning is more complicated with the idempotent/transactional producer capabilities. The following
+ * are the key points:
+ *
+ *
+ * - In order to maintain sequence number continuity for active producers, we always retain the last batch
+ * from each producerId, even if all the records from the batch have been removed. The batch will be removed
+ * once the producer either writes a new batch or is expired due to inactivity.
+ * - We do not clean beyond the last stable offset. This ensures that all records observed by the cleaner have
+ * been decided (i.e. committed or aborted). In particular, this allows us to use the transaction index to
+ * collect the aborted transactions ahead of time.
+ * - Records from aborted transactions are removed by the cleaner immediately without regard to record keys.
+ * - Transaction markers are retained until all record batches from the same transaction have been removed and
+ * a sufficient amount of time has passed to reasonably ensure that an active consumer wouldn't consume any
+ * data from the transaction prior to reaching the offset of the marker. This follows the same logic used for
+ * tombstone deletion.
+ *
+ */
+public class LogCleaner implements BrokerReconfigurable {
+ private static final Logger LOG = LoggerFactory.getLogger(LogCleaner.class);
+
+ public static final Set RECONFIGURABLE_CONFIGS = Set.of(
+ CleanerConfig.LOG_CLEANER_THREADS_PROP,
+ CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP,
+ CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_LOAD_FACTOR_PROP,
+ CleanerConfig.LOG_CLEANER_IO_BUFFER_SIZE_PROP,
+ ServerConfigs.MESSAGE_MAX_BYTES_CONFIG,
+ CleanerConfig.LOG_CLEANER_IO_MAX_BYTES_PER_SECOND_PROP,
+ CleanerConfig.LOG_CLEANER_BACKOFF_MS_PROP
+ );
+
+ // Visible for test
+ public static final String MAX_BUFFER_UTILIZATION_PERCENT_METRIC_NAME = "max-buffer-utilization-percent";
+ public static final String MAX_CLEAN_TIME_METRIC_NAME = "max-clean-time-secs";
+ public static final String MAX_COMPACTION_DELAY_METRICS_NAME = "max-compaction-delay-secs";
+
+ private static final String CLEANER_RECOPY_PERCENT_METRIC_NAME = "cleaner-recopy-percent";
+ private static final String DEAD_THREAD_COUNT_METRIC_NAME = "DeadThreadCount";
+
+ // Visible for test
+ public static final Set METRIC_NAMES = Set.of(
+ MAX_BUFFER_UTILIZATION_PERCENT_METRIC_NAME,
+ CLEANER_RECOPY_PERCENT_METRIC_NAME,
+ MAX_CLEAN_TIME_METRIC_NAME,
+ MAX_COMPACTION_DELAY_METRICS_NAME,
+ DEAD_THREAD_COUNT_METRIC_NAME
+ );
+
+ // For compatibility, metrics are defined to be under `kafka.log.LogCleaner` class
+ private final KafkaMetricsGroup metricsGroup = new KafkaMetricsGroup("kafka.log", "LogCleaner");
+
+ /**
+ * For managing the state of partitions being cleaned.
+ */
+ private final LogCleanerManager cleanerManager;
+
+ /**
+ * A throttle used to limit the I/O of all the cleaner threads to a user-specified maximum rate.
+ */
+ private final Throttler throttler;
+ private final ConcurrentMap logs;
+ private final LogDirFailureChannel logDirFailureChannel;
+ private final Time time;
+ private final List cleaners = new ArrayList<>();
+
+ /**
+ * Log cleaner configuration which may be dynamically updated.
+ */
+ private volatile CleanerConfig config;
+
+ /**
+ * @param initialConfig Initial configuration parameters for the cleaner. Actual config may be dynamically updated.
+ * @param logDirs The directories where offset checkpoints reside
+ * @param logs The map of logs
+ * @param logDirFailureChannel The channel used to add offline log dirs that may be encountered when cleaning the log
+ * @param time A way to control the passage of time
+ */
+ @SuppressWarnings("this-escape")
+ public LogCleaner(CleanerConfig initialConfig,
+ List logDirs,
+ ConcurrentMap logs,
+ LogDirFailureChannel logDirFailureChannel,
+ Time time) {
+ config = initialConfig;
+ this.logs = logs;
+ this.logDirFailureChannel = logDirFailureChannel;
+ cleanerManager = new LogCleanerManager(logDirs, logs, this.logDirFailureChannel);
+ this.time = time;
+ throttler = new Throttler(config.maxIoBytesPerSecond, 300, "cleaner-io", "bytes", this.time);
+
+ registerMetrics();
+ }
+
+ private void registerMetrics() {
+ /* a metric to track the maximum utilization of any thread's buffer in the last cleaning */
+ metricsGroup.newGauge(MAX_BUFFER_UTILIZATION_PERCENT_METRIC_NAME,
+ () -> (int) (maxOverCleanerThreads(t -> t.lastStats.bufferUtilization) * 100));
+
+ /* a metric to track the recopy rate of each thread's last cleaning */
+ metricsGroup.newGauge(CLEANER_RECOPY_PERCENT_METRIC_NAME, () -> {
+ List stats = cleaners.stream().map(t -> t.lastStats).toList();
+ double recopyRate = (double) stats.stream().mapToLong(stat -> stat.bytesWritten).sum() /
+ Math.max(stats.stream().mapToLong(stat -> stat.bytesRead).sum(), 1);
+ return (int) (100 * recopyRate);
+ });
+
+ /* a metric to track the maximum cleaning time for the last cleaning from each thread */
+ metricsGroup.newGauge(MAX_CLEAN_TIME_METRIC_NAME, () -> (int) maxOverCleanerThreads(t -> t.lastStats.elapsedSecs()));
+
+ // a metric to track delay between the time when a log is required to be compacted
+ // as determined by max compaction lag and the time of last cleaner run.
+ metricsGroup.newGauge(MAX_COMPACTION_DELAY_METRICS_NAME,
+ () -> (int) (maxOverCleanerThreads(t -> (double) t.lastPreCleanStats.maxCompactionDelayMs()) / 1000));
+
+ metricsGroup.newGauge(DEAD_THREAD_COUNT_METRIC_NAME, this::deadThreadCount);
+ }
+
+ /**
+ * Start the background cleaner threads.
+ */
+ public void startup() {
+ LOG.info("Starting the log cleaner");
+ IntStream.range(0, config.numThreads).forEach(i -> {
+ try {
+ CleanerThread cleaner = new CleanerThread(i);
+ cleaners.add(cleaner);
+ cleaner.start();
+ } catch (NoSuchAlgorithmException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ /**
+ * Stop the background cleaner threads.
+ */
+ private void shutdownCleaners() {
+ LOG.info("Shutting down the log cleaner.");
+ cleaners.forEach(thread -> {
+ try {
+ thread.shutdown();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ cleaners.clear();
+ }
+
+ /**
+ * Stop the background cleaner threads.
+ */
+ public void shutdown() {
+ try {
+ shutdownCleaners();
+ } finally {
+ removeMetrics();
+ }
+ }
+
+ /**
+ * Remove metrics.
+ */
+ public void removeMetrics() {
+ METRIC_NAMES.forEach(metricsGroup::removeMetric);
+ cleanerManager.removeMetrics();
+ }
+
+ /**
+ * @return A set of configs that is reconfigurable in LogCleaner
+ */
+ @Override
+ public Set reconfigurableConfigs() {
+ return RECONFIGURABLE_CONFIGS;
+ }
+
+ /**
+ * Validate the new cleaner threads num is reasonable.
+ *
+ * @param newConfig A submitted new AbstractConfig instance that contains new cleaner config
+ */
+ @Override
+ public void validateReconfiguration(AbstractConfig newConfig) {
+ int numThreads = new CleanerConfig(newConfig).numThreads;
+ int currentThreads = config.numThreads;
+ if (numThreads < 1)
+ throw new ConfigException("Log cleaner threads should be at least 1");
+ if (numThreads < currentThreads / 2)
+ throw new ConfigException("Log cleaner threads cannot be reduced to less than half the current value " + currentThreads);
+ if (numThreads > currentThreads * 2)
+ throw new ConfigException("Log cleaner threads cannot be increased to more than double the current value " + currentThreads);
+
+ }
+
+ /**
+ * Reconfigure log clean config. The will:
+ *
+ * - update desiredRatePerSec in Throttler with logCleanerIoMaxBytesPerSecond, if necessary
+ * - stop current log cleaners and create new ones.
+ *
+ * That ensures that if any of the cleaners had failed, new cleaners are created to match the new config.
+ *
+ * @param oldConfig the old log cleaner config
+ * @param newConfig the new log cleaner config reconfigured
+ */
+ @Override
+ public void reconfigure(AbstractConfig oldConfig, AbstractConfig newConfig) {
+ config = new CleanerConfig(newConfig);
+
+ double maxIoBytesPerSecond = config.maxIoBytesPerSecond;
+ if (maxIoBytesPerSecond != oldConfig.getDouble(CleanerConfig.LOG_CLEANER_IO_MAX_BYTES_PER_SECOND_PROP)) {
+ LOG.info("Updating logCleanerIoMaxBytesPerSecond: {}", maxIoBytesPerSecond);
+ throttler.updateDesiredRatePerSec(maxIoBytesPerSecond);
+ }
+ // call shutdownCleaners() instead of shutdown to avoid unnecessary deletion of metrics
+ shutdownCleaners();
+ startup();
+ }
+
+ /**
+ * Abort the cleaning of a particular partition, if it's in progress. This call blocks until the cleaning of
+ * the partition is aborted.
+ *
+ * @param topicPartition The topic and partition to abort cleaning
+ */
+ public void abortCleaning(TopicPartition topicPartition) {
+ cleanerManager.abortCleaning(topicPartition);
+ }
+
+ /**
+ * Update checkpoint file to remove partitions if necessary.
+ *
+ * @param dataDir The data dir to be updated if necessary
+ * @param partitionToRemove The topicPartition to be removed
+ */
+ public void updateCheckpoints(File dataDir, Optional partitionToRemove) {
+ cleanerManager.updateCheckpoints(dataDir, Optional.empty(), partitionToRemove);
+ }
+
+ /**
+ * Alter the checkpoint directory for the `topicPartition`, to remove the data in `sourceLogDir`, and add the data in `destLogDir`.
+ * Generally occurs when the disk balance ends and replaces the previous file with the future file.
+ *
+ * @param topicPartition The topic and partition to alter checkpoint
+ * @param sourceLogDir The source log dir to remove checkpoint
+ * @param destLogDir The dest log dir to remove checkpoint
+ */
+ public void alterCheckpointDir(TopicPartition topicPartition, File sourceLogDir, File destLogDir) {
+ cleanerManager.alterCheckpointDir(topicPartition, sourceLogDir, destLogDir);
+ }
+
+ /**
+ * Stop cleaning logs in the provided directory when handling log dir failure.
+ *
+ * @param dir the absolute path of the log dir
+ */
+ public void handleLogDirFailure(String dir) {
+ cleanerManager.handleLogDirFailure(dir);
+ }
+
+ /**
+ * Truncate cleaner offset checkpoint for the given partition if its checkpoint offset is larger than the given offset.
+ *
+ * @param dataDir The data dir to be truncated if necessary
+ * @param topicPartition The topic and partition to truncate checkpoint offset
+ * @param offset The given offset to be compared
+ */
+ public void maybeTruncateCheckpoint(File dataDir, TopicPartition topicPartition, long offset) {
+ cleanerManager.maybeTruncateCheckpoint(dataDir, topicPartition, offset);
+ }
+
+ /**
+ * Abort the cleaning of a particular partition if it's in progress, and pause any future cleaning of this partition.
+ * This call blocks until the cleaning of the partition is aborted and paused.
+ *
+ * @param topicPartition The topic and partition to abort and pause cleaning
+ */
+ public void abortAndPauseCleaning(TopicPartition topicPartition) {
+ cleanerManager.abortAndPauseCleaning(topicPartition);
+ }
+
+ /**
+ * Resume the cleaning of paused partitions.
+ *
+ * @param topicPartitions The collection of topicPartitions to be resumed cleaning
+ */
+ public void resumeCleaning(Set topicPartitions) {
+ cleanerManager.resumeCleaning(topicPartitions);
+ }
+
+ /**
+ * For testing, a way to know when work has completed. This method waits until the
+ * cleaner has processed up to the given offset on the specified topic/partition.
+ *
+ * @param topicPartition The topic and partition to be cleaned
+ * @param offset The first dirty offset that the cleaner doesn't have to clean
+ * @param maxWaitMs The maximum time in ms to wait for cleaner
+ *
+ * @return A boolean indicating whether the work has completed before timeout
+ */
+ public boolean awaitCleaned(TopicPartition topicPartition, long offset, long maxWaitMs) throws InterruptedException {
+ long remainingWaitMs = maxWaitMs;
+ while (!isCleaned(topicPartition, offset) && remainingWaitMs > 0) {
+ long sleepTime = Math.min(100, remainingWaitMs);
+ Thread.sleep(sleepTime);
+ remainingWaitMs -= sleepTime;
+ }
+ return isCleaned(topicPartition, offset);
+ }
+
+ private boolean isCleaned(TopicPartition topicPartition, long offset) {
+ return Optional.ofNullable(cleanerManager.allCleanerCheckpoints().get(topicPartition))
+ .map(checkpoint -> checkpoint >= offset)
+ .orElse(false);
+ }
+
+ /**
+ * To prevent race between retention and compaction,
+ * retention threads need to make this call to obtain:
+ *
+ * @return A map of log partitions that retention threads can safely work on
+ */
+ public Map pauseCleaningForNonCompactedPartitions() {
+ return cleanerManager.pauseCleaningForNonCompactedPartitions().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ }
+
+ /**
+ * @param f to compute the result
+ * @return the max value or 0 if there is no cleaner
+ */
+ public double maxOverCleanerThreads(Function f) {
+ return cleaners.stream()
+ .mapToDouble(f::apply)
+ .max()
+ .orElse(0.0d);
+ }
+
+ public int deadThreadCount() {
+ return (int) cleaners.stream()
+ .filter(ShutdownableThread::isThreadFailed)
+ .count();
+ }
+
+ // Only for testing
+ public LogCleanerManager cleanerManager() {
+ return cleanerManager;
+ }
+
+ // Only for testing
+ public Throttler throttler() {
+ return throttler;
+ }
+
+ // Only for testing
+ public ConcurrentMap logs() {
+ return logs;
+ }
+
+ // Only for testing
+ public List cleaners() {
+ return cleaners;
+ }
+
+ // Only for testing
+ public KafkaMetricsGroup metricsGroup() {
+ return metricsGroup;
+ }
+
+ // Only for testing
+ public CleanerConfig currentConfig() {
+ return config;
+ }
+
+ // Only for testing
+ public int cleanerCount() {
+ return cleaners.size();
+ }
+
+ /**
+ * The cleaner threads do the actual log cleaning. Each thread processes does its cleaning repeatedly by
+ * choosing the dirtiest log, cleaning it, and then swapping in the cleaned segments.
+ */
+ public class CleanerThread extends ShutdownableThread {
+ private final Logger logger = new LogContext(logPrefix).logger(CleanerThread.class);
+
+ private final Cleaner cleaner;
+
+ private volatile CleanerStats lastStats = new CleanerStats(Time.SYSTEM);
+ private volatile PreCleanStats lastPreCleanStats = new PreCleanStats();
+
+ @SuppressWarnings("this-escape")
+ public CleanerThread(int threadId) throws NoSuchAlgorithmException {
+ super("kafka-log-cleaner-thread-" + threadId, false);
+
+ cleaner = new Cleaner(
+ threadId,
+ new SkimpyOffsetMap((int) Math.min(config.dedupeBufferSize / config.numThreads, Integer.MAX_VALUE), config.hashAlgorithm()),
+ config.ioBufferSize / config.numThreads / 2,
+ config.maxMessageSize,
+ config.dedupeBufferLoadFactor,
+ throttler,
+ time,
+ this::checkDone
+ );
+
+ if (config.dedupeBufferSize / config.numThreads > Integer.MAX_VALUE) {
+ logger.warn("Cannot use more than 2G of cleaner buffer space per cleaner thread, ignoring excess buffer space...");
+ }
+ }
+
+ /**
+ * Check if the cleaning for a partition is aborted. If so, throw an exception.
+ *
+ * @param topicPartition The topic and partition to check
+ */
+ private void checkDone(TopicPartition topicPartition) {
+ if (!isRunning()) {
+ throw new ThreadShutdownException();
+ }
+
+ cleanerManager.checkCleaningAborted(topicPartition);
+ }
+
+ /**
+ * The main loop for the cleaner thread.
+ * Clean a log if there is a dirty log available, otherwise sleep for a bit.
+ */
+ @Override
+ public void doWork() {
+ boolean cleaned = tryCleanFilthiestLog();
+ if (!cleaned) {
+ try {
+ pause(config.backoffMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ cleanerManager.maintainUncleanablePartitions();
+ }
+
+ public CleanerStats lastStats() {
+ return lastStats;
+ }
+
+ public void setLastStats(CleanerStats lastStats) {
+ this.lastStats = lastStats;
+ }
+
+ public PreCleanStats lastPreCleanStats() {
+ return lastPreCleanStats;
+ }
+
+ /**
+ * Cleans a log if there is a dirty log available.
+ *
+ * @return whether a log was cleaned
+ */
+ private boolean tryCleanFilthiestLog() {
+ try {
+ return cleanFilthiestLog();
+ } catch (LogCleaningException e) {
+ logger.warn("Unexpected exception thrown when cleaning log {}. Marking its partition ({}) as uncleanable", e.log, e.log.topicPartition(), e);
+ cleanerManager.markPartitionUncleanable(e.log.parentDir(), e.log.topicPartition());
+
+ return false;
+ }
+ }
+
+ private boolean cleanFilthiestLog() throws LogCleaningException {
+ PreCleanStats preCleanStats = new PreCleanStats();
+ Optional ltc = cleanerManager.grabFilthiestCompactedLog(time, preCleanStats);
+ boolean cleaned;
+
+ if (ltc.isEmpty()) {
+ cleaned = false;
+ } else {
+ // there's a log, clean it
+ this.lastPreCleanStats = preCleanStats;
+ LogToClean cleanable = null;
+ try {
+ cleanable = ltc.get();
+ cleanLog(cleanable);
+ cleaned = true;
+ } catch (ThreadShutdownException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new LogCleaningException(cleanable != null ? cleanable.log() : null, e.getMessage(), e);
+ }
+ }
+
+ Map deletable = cleanerManager.deletableLogs();
+ try {
+ deletable.forEach((topicPartition, log) -> {
+ try {
+ log.deleteOldSegments();
+ } catch (ThreadShutdownException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new LogCleaningException(log, e.getMessage(), e);
+ }
+ });
+ } finally {
+ cleanerManager.doneDeleting(deletable.keySet().stream().toList());
+ }
+
+ return cleaned;
+ }
+
+ private void cleanLog(LogToClean cleanable) throws DigestException {
+ long startOffset = cleanable.firstDirtyOffset();
+ long endOffset = startOffset;
+ try {
+ Map.Entry entry = cleaner.clean(cleanable);
+ endOffset = entry.getKey();
+ recordStats(cleaner.id(), cleanable.log().name(), startOffset, endOffset, entry.getValue());
+ } catch (LogCleaningAbortedException ignored) {
+ // task can be aborted, let it go.
+ } catch (KafkaStorageException ignored) {
+ // partition is already offline. let it go.
+ } catch (IOException e) {
+ String logDirectory = cleanable.log().parentDir();
+ String msg = String.format("Failed to clean up log for %s in dir %s due to IOException", cleanable.topicPartition(), logDirectory);
+ logDirFailureChannel.maybeAddOfflineLogDir(logDirectory, msg, e);
+ } finally {
+ cleanerManager.doneCleaning(cleanable.topicPartition(), cleanable.log().parentDirFile(), endOffset);
+ }
+ }
+
+ /**
+ * Log out statistics on a single run of the cleaner.
+ *
+ * @param id The cleaner thread id
+ * @param name The cleaned log name
+ * @param from The cleaned offset that is the first dirty offset to begin
+ * @param to The cleaned offset that is the first not cleaned offset to end
+ * @param stats The statistics for this round of cleaning
+ */
+ private void recordStats(int id, String name, long from, long to, CleanerStats stats) {
+ this.lastStats = stats;
+ String message = String.format("%n\tLog cleaner thread %d cleaned log %s (dirty section = [%d, %d])%n", id, name, from, to) +
+ String.format("\t%,.1f MB of log processed in %,.1f seconds (%,.1f MB/sec).%n", mb(stats.bytesRead),
+ stats.elapsedSecs(),
+ mb(stats.bytesRead / stats.elapsedSecs())) +
+ String.format("\tIndexed %,.1f MB in %.1f seconds (%,.1f Mb/sec, %.1f%% of total time)%n", mb(stats.mapBytesRead),
+ stats.elapsedIndexSecs(),
+ mb(stats.mapBytesRead) / stats.elapsedIndexSecs(),
+ 100 * stats.elapsedIndexSecs() / stats.elapsedSecs()) +
+ String.format("\tBuffer utilization: %.1f%%%n", 100 * stats.bufferUtilization) +
+ String.format("\tCleaned %,.1f MB in %.1f seconds (%,.1f Mb/sec, %.1f%% of total time)%n", mb(stats.bytesRead),
+ stats.elapsedSecs() - stats.elapsedIndexSecs(),
+ mb(stats.bytesRead) / (stats.elapsedSecs() - stats.elapsedIndexSecs()), 100 * (stats.elapsedSecs() - stats.elapsedIndexSecs()) / stats.elapsedSecs()) +
+ String.format("\tStart size: %,.1f MB (%,d messages)%n", mb(stats.bytesRead), stats.messagesRead) +
+ String.format("\tEnd size: %,.1f MB (%,d messages)%n", mb(stats.bytesWritten), stats.messagesWritten) +
+ String.format("\t%.1f%% size reduction (%.1f%% fewer messages)%n", 100.0 * (1.0 - Long.valueOf(stats.bytesWritten).doubleValue() / stats.bytesRead),
+ 100.0 * (1.0 - Long.valueOf(stats.messagesWritten).doubleValue() / stats.messagesRead));
+ logger.info(message);
+ if (lastPreCleanStats.delayedPartitions() > 0) {
+ logger.info("\tCleanable partitions: {}, Delayed partitions: {}, max delay: {}",
+ lastPreCleanStats.cleanablePartitions(), lastPreCleanStats.delayedPartitions(), lastPreCleanStats.maxCompactionDelayMs());
+ }
+ if (stats.invalidMessagesRead > 0) {
+ logger.warn("\tFound {} invalid messages during compaction.", stats.invalidMessagesRead);
+ }
+ }
+
+ private double mb(double bytes) {
+ return bytes / (1024 * 1024);
+ }
+ }
+}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleanerManager.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleanerManager.java
index dfee694afefb4..0b8c2b398d039 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleanerManager.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleanerManager.java
@@ -346,7 +346,7 @@ public Map deletableLogs() {
public void abortCleaning(TopicPartition topicPartition) {
inLock(lock, () -> {
abortAndPauseCleaning(topicPartition);
- resumeCleaning(List.of(topicPartition));
+ resumeCleaning(Set.of(topicPartition));
return null;
});
}
@@ -396,7 +396,7 @@ public void abortAndPauseCleaning(TopicPartition topicPartition) {
* Resume the cleaning of paused partitions.
* Each call of this function will undo one pause.
*/
- public void resumeCleaning(List topicPartitions) {
+ public void resumeCleaning(Set topicPartitions) {
inLock(lock, () -> {
topicPartitions.forEach(topicPartition -> {
LogCleaningState state = inProgress.get(topicPartition);
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
index 2e89cdee21d40..21c92cd84dff4 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
@@ -27,7 +27,6 @@
import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.record.LegacyRecord;
-import org.apache.kafka.common.record.Records;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.utils.ConfigUtils;
import org.apache.kafka.common.utils.Utils;
@@ -124,7 +123,6 @@ public Optional serverConfigName(String configName) {
// Visible for testing
public static final String SERVER_DEFAULT_HEADER_NAME = "Server Default Property";
- public static final int DEFAULT_MAX_MESSAGE_BYTES = 1024 * 1024 + Records.LOG_OVERHEAD;
public static final int DEFAULT_SEGMENT_BYTES = 1024 * 1024 * 1024;
public static final long DEFAULT_SEGMENT_MS = 24 * 7 * 60 * 60 * 1000L;
public static final long DEFAULT_SEGMENT_JITTER_MS = 0;
@@ -134,7 +132,6 @@ public Optional serverConfigName(String configName) {
public static final long DEFAULT_MAX_COMPACTION_LAG_MS = Long.MAX_VALUE;
public static final double DEFAULT_MIN_CLEANABLE_DIRTY_RATIO = 0.5;
public static final boolean DEFAULT_UNCLEAN_LEADER_ELECTION_ENABLE = false;
- public static final String DEFAULT_COMPRESSION_TYPE = BrokerCompressionType.PRODUCER.name;
public static final boolean DEFAULT_PREALLOCATE = false;
public static final boolean DEFAULT_REMOTE_STORAGE_ENABLE = false;
@@ -210,7 +207,7 @@ public Optional serverConfigName(String configName) {
// can be negative. See kafka.log.LogManager.cleanupExpiredSegments
.define(TopicConfig.RETENTION_MS_CONFIG, LONG, DEFAULT_RETENTION_MS, atLeast(-1), MEDIUM,
TopicConfig.RETENTION_MS_DOC)
- .define(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, INT, DEFAULT_MAX_MESSAGE_BYTES, atLeast(0), MEDIUM,
+ .define(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, INT, ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT, atLeast(0), MEDIUM,
TopicConfig.MAX_MESSAGE_BYTES_DOC)
.define(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, INT, ServerLogConfigs.LOG_INDEX_INTERVAL_BYTES_DEFAULT, atLeast(0), MEDIUM,
TopicConfig.INDEX_INTERVAL_BYTES_DOC)
@@ -230,7 +227,7 @@ public Optional serverConfigName(String configName) {
MEDIUM, TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_DOC)
.define(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, INT, ServerLogConfigs.MIN_IN_SYNC_REPLICAS_DEFAULT, atLeast(1), MEDIUM,
TopicConfig.MIN_IN_SYNC_REPLICAS_DOC)
- .define(TopicConfig.COMPRESSION_TYPE_CONFIG, STRING, DEFAULT_COMPRESSION_TYPE, in(BrokerCompressionType.names().toArray(new String[0])),
+ .define(TopicConfig.COMPRESSION_TYPE_CONFIG, STRING, ServerLogConfigs.COMPRESSION_TYPE_DEFAULT, in(BrokerCompressionType.names().toArray(new String[0])),
MEDIUM, TopicConfig.COMPRESSION_TYPE_DOC)
.define(TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG, INT, CompressionType.GZIP.defaultLevel(),
CompressionType.GZIP.levelValidator(), MEDIUM, TopicConfig.COMPRESSION_GZIP_LEVEL_DOC)
diff --git a/tests/kafkatest/services/kafka/templates/log4j2.yaml b/tests/kafkatest/services/kafka/templates/log4j2.yaml
index 22e3f118f680d..ff2c01e4c1be5 100644
--- a/tests/kafkatest/services/kafka/templates/log4j2.yaml
+++ b/tests/kafkatest/services/kafka/templates/log4j2.yaml
@@ -254,7 +254,21 @@ Configuration:
- ref: ControllerInfoAppender
- ref: ControllerDebugAppender
- - name: kafka.log.LogCleaner
+ - name: org.apache.kafka.storage.internals.log.LogCleaner
+ level: {{ log_level|default("DEBUG") }}
+ additivity: false
+ AppenderRef:
+ - ref: CleanerInfoAppender
+ - ref: CleanerDebugAppender
+
+ - name: org.apache.kafka.storage.internals.log.LogCleaner$CleanerThread
+ level: {{ log_level|default("DEBUG") }}
+ additivity: false
+ AppenderRef:
+ - ref: CleanerInfoAppender
+ - ref: CleanerDebugAppender
+
+ - name: org.apache.kafka.storage.internals.log.Cleaner
level: {{ log_level|default("DEBUG") }}
additivity: false
AppenderRef: