You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2024/04/24 09:51:36 UTC
(kafka) branch trunk updated: KAFKA-16424: remove truncated logs after alter dir (#15616)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a38185280ca KAFKA-16424: remove truncated logs after alter dir (#15616)
a38185280ca is described below
commit a38185280ca011d4caced05171e94ac53cda964d
Author: PoAn Yang <pa...@apache.org>
AuthorDate: Wed Apr 24 17:51:29 2024 +0800
KAFKA-16424: remove truncated logs after alter dir (#15616)
If there are some logs to be deleted during the log dir movement, we'll send for a scheduler to do the deletion later.
However, when the log dir movement completed, the future log is renamed, the async log deletion will fail with no file existed error.
Signed-off-by: PoAn Yang <pa...@apache.org>
Reviewers: Luke Chen <sh...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>, SoontaekLim <so...@neya.kr>, Johnny Hsu <jo...@fb.com>
---
.../server/AlterReplicaLogDirsRequestTest.scala | 62 +++++++++++++++++++++-
.../kafka/storage/internals/log/LogSegment.java | 23 +++++++-
2 files changed, 82 insertions(+), 3 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala b/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
index 2653d3f9656..4574114ee7d 100644
--- a/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
@@ -18,15 +18,18 @@
package kafka.server
import java.io.File
-
import kafka.utils._
import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.message.AlterReplicaLogDirsRequestData
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{AlterReplicaLogDirsRequest, AlterReplicaLogDirsResponse}
+import org.apache.kafka.server.config.ServerLogConfigs
+import org.apache.kafka.storage.internals.log.LogFileUtils
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
+import java.util.Properties
import scala.jdk.CollectionConverters._
import scala.collection.mutable
import scala.util.Random
@@ -37,6 +40,11 @@ class AlterReplicaLogDirsRequestTest extends BaseRequestTest {
val topic = "topic"
+ override def brokerPropertyOverrides(properties: Properties): Unit = {
+ properties.put(ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_CONFIG, "0")
+ properties.put(ServerLogConfigs.LOG_CLEANUP_INTERVAL_MS_CONFIG, "1000")
+ }
+
private def findErrorForPartition(response: AlterReplicaLogDirsResponse, tp: TopicPartition): Errors = {
Errors.forCode(response.data.results.asScala
.find(x => x.topicName == tp.topic).get.partitions.asScala
@@ -116,6 +124,58 @@ class AlterReplicaLogDirsRequestTest extends BaseRequestTest {
assertEquals(Errors.KAFKA_STORAGE_ERROR, findErrorForPartition(alterReplicaDirResponse3, new TopicPartition(topic, 2)))
}
+ @Test
+ def testAlterReplicaLogDirsRequestWithRetention(): Unit = {
+ val partitionNum = 1
+
+ // Alter replica dir before topic creation
+ val logDir1 = new File(servers.head.config.logDirs(1)).getAbsolutePath
+ val partitionDirs1 = (0 until partitionNum).map(partition => new TopicPartition(topic, partition) -> logDir1).toMap
+ val alterReplicaLogDirsResponse1 = sendAlterReplicaLogDirsRequest(partitionDirs1)
+
+ // The response should show error UNKNOWN_TOPIC_OR_PARTITION for all partitions
+ val tp = new TopicPartition(topic, 0)
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, findErrorForPartition(alterReplicaLogDirsResponse1, tp))
+ assertTrue(servers.head.logManager.getLog(tp).isEmpty)
+
+ val topicProperties = new Properties()
+ topicProperties.put(TopicConfig.RETENTION_BYTES_CONFIG, "1024")
+ // This test needs enough time to wait for dir movement happened.
+ // We don't want files with `.deleted` suffix are removed too fast,
+ // so we can validate there will be orphan files and orphan files will be removed eventually.
+ topicProperties.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, "10000")
+ topicProperties.put(TopicConfig.SEGMENT_BYTES_CONFIG, "1024")
+
+ createTopic(topic, partitionNum, 1, topicProperties)
+ assertEquals(logDir1, servers.head.logManager.getLog(tp).get.dir.getParent)
+
+ // send enough records to trigger log rolling
+ (0 until 20).foreach { _ =>
+ TestUtils.generateAndProduceMessages(servers, topic, 10, 1)
+ }
+ TestUtils.waitUntilTrue(() => servers.head.logManager.getLog(new TopicPartition(topic, 0)).get.numberOfSegments > 1,
+ "timed out waiting for log segment to roll")
+
+ // Wait for log segment retention in original dir.
+ TestUtils.waitUntilTrue(() => {
+ new File(logDir1, tp.toString).listFiles().count(_.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)) > 0
+ }, "timed out waiting for log segment to retention")
+
+ // Alter replica dir again after topic creation
+ val logDir2 = new File(servers.head.config.logDirs(2)).getAbsolutePath
+ val alterReplicaLogDirsResponse2 = sendAlterReplicaLogDirsRequest(Map(tp -> logDir2))
+ // The response should succeed for all partitions
+ assertEquals(Errors.NONE, findErrorForPartition(alterReplicaLogDirsResponse2, tp))
+ TestUtils.waitUntilTrue(() => {
+ logDir2 == servers.head.logManager.getLog(tp).get.dir.getParent
+ }, "timed out waiting for replica movement")
+
+ // Make sure the deleted log segment is removed
+ TestUtils.waitUntilTrue(() => {
+ new File(logDir2, tp.toString).listFiles().count(_.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)) == 0
+ }, "timed out waiting for removing deleted log segment")
+ }
+
private def sendAlterReplicaLogDirsRequest(partitionDirs: Map[TopicPartition, String]): AlterReplicaLogDirsResponse = {
val logDirs = partitionDirs.groupBy{case (_, dir) => dir}.map{ case(dir, tps) =>
new AlterReplicaLogDirsRequestData.AlterReplicaLogDir()
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
index 2f806bfcf32..104eb9a18b3 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
@@ -29,6 +29,8 @@ import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import com.yammer.metrics.core.MetricName;
import com.yammer.metrics.core.Timer;
@@ -63,6 +65,9 @@ import static java.util.Arrays.asList;
public class LogSegment implements Closeable {
private static final Logger LOGGER = LoggerFactory.getLogger(LogSegment.class);
private static final Timer LOG_FLUSH_TIMER;
+ /* a directory that is used for future partition */
+ private static final String FUTURE_DIR_SUFFIX = "-future";
+ private static final Pattern FUTURE_DIR_PATTERN = Pattern.compile("^(\\S+)-(\\S+)\\.(\\S+)" + FUTURE_DIR_SUFFIX);
static {
KafkaMetricsGroup logFlushStatsMetricsGroup = new KafkaMetricsGroup(LogSegment.class) {
@@ -801,8 +806,22 @@ public class LogSegment implements Closeable {
try {
if (delete.execute())
LOGGER.info("Deleted {} {}.", fileType, file.getAbsolutePath());
- else if (logIfMissing)
- LOGGER.info("Failed to delete {} {} because it does not exist.", fileType, file.getAbsolutePath());
+ else {
+ if (logIfMissing) {
+ LOGGER.info("Failed to delete {} {} because it does not exist.", fileType, file.getAbsolutePath());
+ }
+
+ // During alter log dir, the log segment may be moved to a new directory, so async delete may fail.
+ // Fallback to delete the file in the new directory to avoid orphan file.
+ Matcher dirMatcher = FUTURE_DIR_PATTERN.matcher(file.getParent());
+ if (dirMatcher.matches()) {
+ String topicPartitionAbsolutePath = dirMatcher.group(1) + "-" + dirMatcher.group(2);
+ File fallbackFile = new File(topicPartitionAbsolutePath, file.getName());
+ if (fallbackFile.exists() && file.getName().endsWith(LogFileUtils.DELETED_FILE_SUFFIX) && fallbackFile.delete()) {
+ LOGGER.info("Fallback to delete {} {}.", fileType, fallbackFile.getAbsolutePath());
+ }
+ }
+ }
return null;
} catch (IOException e) {
throw new IOException("Delete of " + fileType + " " + file.getAbsolutePath() + " failed.", e);