Skip to content

Commit 02e40ec

Browse files
gyang94gyang94
authored
fix: remote orphan files and directories when tablet starts up (#3388)
1 parent 91bea7c commit 02e40ec

4 files changed

Lines changed: 316 additions & 15 deletions

File tree

‎fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java‎

Lines changed: 56 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.fluss.exception.FlussException;
2424
import org.apache.fluss.exception.FlussRuntimeException;
2525
import org.apache.fluss.exception.LogStorageException;
26+
import org.apache.fluss.exception.PartitionNotExistException;
2627
import org.apache.fluss.exception.SchemaNotExistException;
2728
import org.apache.fluss.metadata.LogFormat;
2829
import org.apache.fluss.metadata.PhysicalTablePath;
@@ -34,6 +35,7 @@
3435
import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
3536
import org.apache.fluss.server.storage.LocalDiskManager;
3637
import org.apache.fluss.server.zk.ZooKeeperClient;
38+
import org.apache.fluss.server.zk.data.PartitionRegistration;
3739
import org.apache.fluss.utils.ExceptionUtils;
3840
import org.apache.fluss.utils.FileUtils;
3941
import org.apache.fluss.utils.FlussPaths;
@@ -48,7 +50,9 @@
4850

4951
import java.io.File;
5052
import java.io.IOException;
53+
import java.nio.file.DirectoryNotEmptyException;
5154
import java.nio.file.Files;
55+
import java.nio.file.NoSuchFileException;
5256
import java.util.ArrayList;
5357
import java.util.HashMap;
5458
import java.util.LinkedHashMap;
@@ -386,6 +390,25 @@ private LogTablet loadLog(
386390
PhysicalTablePath physicalTablePath = pathAndBucket.f0;
387391
TablePath tablePath = physicalTablePath.getTablePath();
388392
TableInfo tableInfo = getTableInfo(zkClient, tablePath);
393+
394+
// Table schema exists, but the partition may have been dropped (or dropped
395+
// and recreated with a new ID) while this TS was offline.
396+
// Validate both partition name and partition ID against ZK.
397+
String partitionName = physicalTablePath.getPartitionName();
398+
if (partitionName != null) {
399+
Optional<PartitionRegistration> registration =
400+
zkClient.getPartition(tablePath, partitionName);
401+
if (!registration.isPresent()
402+
|| registration.get().getPartitionId() != tableBucket.getPartitionId()) {
403+
throw new PartitionNotExistException(
404+
String.format(
405+
"Failed to load partition '%s' (partitionId=%d) of table '%s': "
406+
+ "partition not found or partitionId mismatch in "
407+
+ "zookeeper metadata.",
408+
partitionName, tableBucket.getPartitionId(), tablePath));
409+
}
410+
}
411+
389412
LogTablet logTablet =
390413
LogTablet.create(
391414
dataDir,
@@ -535,17 +558,19 @@ public void run() {
535558
loadLog(dataDir, tabletDir, cleanShutdown, recoveryPoints, conf, clock);
536559
} catch (Exception e) {
537560
LOG.error("Fail to loadLog from {}", tabletDir, e);
538-
if (e instanceof SchemaNotExistException) {
561+
if (e instanceof SchemaNotExistException
562+
|| e instanceof PartitionNotExistException) {
539563
LOG.error(
540-
"schema not exist, table for {} has already been dropped, the residual data will be removed.",
564+
"Table or partition for {} has already been dropped, the residual data will be removed.",
541565
tabletDir,
542566
e);
543567
FileUtils.deleteDirectoryQuietly(tabletDir);
544568

545-
// Also delete corresponding KV tablet directory if it exists
546569
try {
547570
Tuple2<PhysicalTablePath, TableBucket> pathAndBucket =
548571
FlussPaths.parseTabletDir(tabletDir);
572+
573+
// Also delete corresponding KV tablet directory if it exists
549574
File kvTabletDir =
550575
FlussPaths.kvTabletDir(
551576
dataDir, pathAndBucket.f0, pathAndBucket.f1);
@@ -555,11 +580,24 @@ public void run() {
555580
kvTabletDir);
556581
FileUtils.deleteDirectoryQuietly(kvTabletDir);
557582
}
558-
} catch (Exception kvDeleteException) {
583+
584+
boolean isPartitioned = pathAndBucket.f0.getPartitionName() != null;
585+
File partitionDir = tabletDir.getParentFile();
586+
if (partitionDir != null) {
587+
deleteEmptyDirQuietly(partitionDir);
588+
589+
if (isPartitioned) {
590+
File tableDir = partitionDir.getParentFile();
591+
if (tableDir != null) {
592+
deleteEmptyDirQuietly(tableDir);
593+
}
594+
}
595+
}
596+
} catch (Exception cleanupException) {
559597
LOG.warn(
560-
"Failed to delete corresponding KV tablet directory for log {}: {}",
598+
"Failed to clean up residual KV/parent directories for {}: {}",
561599
tabletDir,
562-
kvDeleteException.getMessage());
600+
cleanupException.getMessage());
563601
}
564602
return;
565603
}
@@ -607,6 +645,18 @@ private LogRecoveryTask(
607645
}
608646
}
609647

648+
private static void deleteEmptyDirQuietly(File dir) {
649+
try {
650+
Files.delete(dir.toPath());
651+
} catch (DirectoryNotEmptyException e) {
652+
LOG.warn("Directory {} is not empty, skipping deletion.", dir);
653+
} catch (NoSuchFileException ignored) {
654+
// Already gone — fine.
655+
} catch (IOException e) {
656+
LOG.warn("Failed to delete empty directory {}: {}", dir, e.getMessage());
657+
}
658+
}
659+
610660
private static final class LogShutdownTask {
611661
private final File dataDir;
612662
private final List<LogTablet> logs;

‎fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java‎

Lines changed: 82 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.fluss.exception.InvalidColumnProjectionException;
2828
import org.apache.fluss.exception.InvalidCoordinatorException;
2929
import org.apache.fluss.exception.InvalidRequiredAcksException;
30+
import org.apache.fluss.exception.KvStorageException;
3031
import org.apache.fluss.exception.LogOffsetOutOfRangeException;
3132
import org.apache.fluss.exception.LogStorageException;
3233
import org.apache.fluss.exception.NotLeaderOrFollowerException;
@@ -124,7 +125,9 @@
124125

125126
import java.io.File;
126127
import java.io.IOException;
128+
import java.nio.file.DirectoryNotEmptyException;
127129
import java.nio.file.Files;
130+
import java.nio.file.NoSuchFileException;
128131
import java.nio.file.Path;
129132
import java.util.ArrayList;
130133
import java.util.Collections;
@@ -145,7 +148,6 @@
145148

146149
import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
147150
import static org.apache.fluss.server.TabletManagerBase.getTableInfo;
148-
import static org.apache.fluss.utils.FileUtils.isDirectoryEmpty;
149151
import static org.apache.fluss.utils.Preconditions.checkArgument;
150152
import static org.apache.fluss.utils.Preconditions.checkNotNull;
151153
import static org.apache.fluss.utils.Preconditions.checkState;
@@ -978,7 +980,20 @@ public void stopReplicas(
978980
TableBucket tb = data.getTableBucket();
979981
HostedReplica hostedReplica = getReplica(tb);
980982
if (hostedReplica instanceof NoneReplica) {
981-
// do nothing fort this case.
983+
if (data.isDeleteLocal()) {
984+
try {
985+
sweepOrphanTabletDirs(tb, deletedTableIds, deletedPartitionIds);
986+
} catch (Exception e) {
987+
LOG.error(
988+
"Failed to sweep orphan tablet directories for {}",
989+
tb,
990+
e);
991+
result.add(
992+
new StopReplicaResultForBucket(
993+
tb, ApiError.fromThrowable(e)));
994+
continue;
995+
}
996+
}
982997
result.add(new StopReplicaResultForBucket(tb));
983998
} else if (hostedReplica instanceof OfflineReplica) {
984999
LOG.warn(
@@ -1927,6 +1942,59 @@ private StopReplicaResultForBucket stopReplica(
19271942
return new StopReplicaResultForBucket(tb);
19281943
}
19291944

1945+
/**
1946+
* Remove on-disk tablet directories for a bucket that the in-memory ReplicaManager does not
1947+
* know about. This handles the case where a stopReplica(delete=true) arrives after the
1948+
* TabletServer was restarted during a delete — LogManager loaded the log at startup but no
1949+
* NotifyLeaderAndIsr ever ran, so allReplicas is empty.
1950+
*/
1951+
private void sweepOrphanTabletDirs(
1952+
TableBucket tb, Map<Long, Path> deletedTableIds, Map<Long, Path> deletedPartitionIds) {
1953+
Optional<LogTablet> orphanLog = logManager.getLog(tb);
1954+
if (!orphanLog.isPresent()) {
1955+
return;
1956+
}
1957+
1958+
LogTablet logTablet = orphanLog.get();
1959+
File dataDir = logTablet.getDataDir();
1960+
PhysicalTablePath physicalTablePath = logTablet.getPhysicalTablePath();
1961+
Path tabletParentDir = logManager.getTabletParentDir(dataDir, physicalTablePath, tb);
1962+
1963+
// Clean KV before log so that if KV cleanup fails, the log is still
1964+
// present and a coordinator retry can re-enter this method.
1965+
boolean isKvTable = false;
1966+
if (kvManager.getKv(tb).isPresent()) {
1967+
kvManager.dropKv(tb);
1968+
isKvTable = true;
1969+
} else {
1970+
File kvTabletDir = FlussPaths.kvTabletDir(dataDir, physicalTablePath, tb);
1971+
if (kvTabletDir.exists()) {
1972+
isKvTable = true;
1973+
try {
1974+
FileUtils.deleteDirectory(kvTabletDir);
1975+
} catch (IOException e) {
1976+
throw new KvStorageException(
1977+
String.format(
1978+
"Failed to delete orphan KV tablet directory %s", kvTabletDir),
1979+
e);
1980+
}
1981+
}
1982+
}
1983+
1984+
logManager.dropLog(tb);
1985+
1986+
localDiskManager.recordReplicaDelete(dataDir, isKvTable);
1987+
1988+
if (tb.getPartitionId() != null) {
1989+
deletedPartitionIds.put(tb.getPartitionId(), tabletParentDir);
1990+
deletedTableIds.put(tb.getTableId(), tabletParentDir.getParent());
1991+
} else {
1992+
deletedTableIds.put(tb.getTableId(), tabletParentDir);
1993+
}
1994+
1995+
LOG.info("Swept orphan tablet directories for bucket {}", tb);
1996+
}
1997+
19301998
private void truncateToHighWatermark(List<Replica> replicas) {
19311999
for (Replica replica : replicas) {
19322000
long highWatermark = replica.getLogTablet().getHighWatermark();
@@ -1960,14 +2028,19 @@ private void validateAndApplyCoordinatorEpoch(int requestCoordinatorEpoch, Strin
19602028
}
19612029

19622030
private void dropEmptyTableOrPartitionDir(Path dir, long id, String dirType) {
1963-
if (!Files.exists(dir) || !isDirectoryEmpty(dir)) {
1964-
return;
1965-
}
1966-
1967-
LOG.info("Drop empty {} dir '{}' of {} id {}.", dirType, dir, dirType, id);
19682031
try {
1969-
FileUtils.deleteDirectory(dir.toFile());
1970-
} catch (Exception e) {
2032+
Files.delete(dir);
2033+
LOG.info("Dropped empty {} dir '{}' of {} id {}.", dirType, dir, dirType, id);
2034+
} catch (DirectoryNotEmptyException e) {
2035+
LOG.warn(
2036+
"{} dir '{}' of {} id {} is not empty, skipping deletion.",
2037+
dirType,
2038+
dir,
2039+
dirType,
2040+
id);
2041+
} catch (NoSuchFileException ignored) {
2042+
// Already gone — fine.
2043+
} catch (IOException e) {
19712044
LOG.error("Failed to delete empty {} dir '{}' of {} id {}.", dirType, dir, dirType, e);
19722045
}
19732046
}

0 commit comments

Comments
 (0)