yihua commented on code in PR #13886:
URL: https://github.com/apache/hudi/pull/13886#discussion_r2392855525
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/LockAuditingCommand.java:
##########
@@ -185,4 +272,346 @@ public String showLockAuditStatus() {
return String.format("Failed to check lock audit status: %s",
e.getMessage());
}
}
+
+ /**
+ * Validates the audit lock files for consistency and integrity.
+ * This command checks for issues such as corrupted files, invalid format,
+ * incorrect state transitions, and orphaned locks.
+ *
+ * @return Validation results including any issues found
+ */
+ @ShellMethod(key = "locks audit validate", value = "Validate audit lock
files for consistency and integrity")
+ public String validateAuditLocks() {
+
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ try {
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit folder found - nothing to validate";
+ }
+
+ // Get all audit files
+ List<StoragePathInfo> allFiles =
HoodieCLI.storage.listDirectEntries(auditFolder);
+ List<StoragePathInfo> auditFiles = new ArrayList<>();
+ for (StoragePathInfo pathInfo : allFiles) {
+ if (pathInfo.isFile() &&
pathInfo.getPath().getName().endsWith(".jsonl")) {
+ auditFiles.add(pathInfo);
+ }
+ }
+
+ if (auditFiles.isEmpty()) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit files found - nothing to validate";
+ }
+
+ // Parse all audit files into transaction windows
+ List<TransactionWindow> windows = new ArrayList<>();
+ for (StoragePathInfo pathInfo : auditFiles) {
+ Option<TransactionWindow> window = parseAuditFile(pathInfo);
+ if (window.isPresent()) {
+ windows.add(window.get());
+ }
+ }
+
+ if (windows.isEmpty()) {
+ return String.format("Validation Result: FAILED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: %d\n"
+ + "Details: Failed to parse any audit files", auditFiles.size());
+ }
+
+ // Validate transactions
+ ValidationResults validationResults =
validateTransactionWindows(windows);
+
+ // Generate result
+ int totalIssues = validationResults.errors.size() +
validationResults.warnings.size();
+ String result;
+ String details;
+
+ if (totalIssues == 0) {
+ result = "PASSED";
+ details = "All audit lock transactions validated successfully";
+ } else {
+ result = validationResults.errors.isEmpty() ? "WARNING" : "FAILED";
+ List<String> allIssues = new ArrayList<>();
+ allIssues.addAll(validationResults.errors);
+ allIssues.addAll(validationResults.warnings);
+ details = String.join(", ", allIssues);
+ }
+
+ return String.format("Validation Result: %s\n"
+ + "Transactions Validated: %d\n"
+ + "Issues Found: %d\n"
+ + "Details: %s", result, windows.size(), totalIssues, details);
+
+ } catch (Exception e) {
+ LOG.error("Error validating audit locks", e);
+ return String.format("Validation Result: ERROR\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: -1\n"
+ + "Details: Validation failed: %s", e.getMessage());
+ }
+ }
+
+ /**
+ * Cleans up old audit lock files based on age threshold.
+ * This command removes audit files that are older than the specified number
of days.
+ *
+ * @param dryRun Whether to perform a dry run (preview changes without
deletion)
+ * @param ageDays Number of days to keep audit files (default 7)
+ * @return Status message indicating files cleaned or to be cleaned
+ */
+ @ShellMethod(key = "locks audit cleanup", value = "Clean up old audit lock
files")
+ public String cleanupAuditLocks(
+ @ShellOption(value = {"--dryRun"}, defaultValue = "false",
+ help = "Preview changes without actually deleting files") final
boolean dryRun,
+ @ShellOption(value = {"--ageDays"}, defaultValue = "7",
+ help = "Delete audit files older than this many days") final String
ageDaysStr) {
+
+ try {
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ // Parse ageDays manually to handle validation properly
+ int ageDays;
+ try {
+ ageDays = Integer.parseInt(ageDaysStr);
+ } catch (NumberFormatException e) {
+ return "Error: ageDays must be a value greater than 0.";
+ }
+
+ if (ageDays < 0) {
+ return "Error: ageDays must be non-negative (>= 0).";
+ }
+
+ return performAuditCleanup(dryRun, ageDays).toString();
+ } catch (Exception e) {
+ LOG.error("Error during audit cleanup", e);
+ return String.format("Error during cleanup: %s", e.getMessage() != null
? e.getMessage() : e.getClass().getSimpleName());
+ }
+ }
+
+ /**
+ * Internal method to perform audit cleanup. Used by both the CLI command
and disable method.
+ *
+ * @param dryRun Whether to perform a dry run (preview changes without
deletion)
+ * @param ageDays Number of days to keep audit files (0 means delete all)
+ * @return CleanupResult containing cleanup operation details
+ */
+ private CleanupResult performAuditCleanup(boolean dryRun, int ageDays) {
+ if (ageDays < 0) {
+ String message = "Error: ageDays must be non-negative (>= 0).";
+ return new CleanupResult(false, 0, 0, 0, message, dryRun);
+ }
+
+ try {
+ if (HoodieCLI.storage == null) {
+ String message = "Storage not initialized.";
+ return new CleanupResult(false, 0, 0, 0, message, dryRun);
+ }
+
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ String message = "No audit folder found - nothing to cleanup.";
+ return new CleanupResult(true, 0, 0, 0, message, dryRun);
+ }
+
+ // Calculate cutoff timestamp (ageDays ago)
+ long cutoffTime = System.currentTimeMillis() - (ageDays * 24L * 60L *
60L * 1000L);
+
+ // List all files in audit folder and filter by modification time
+ List<StoragePathInfo> allFiles =
HoodieCLI.storage.listDirectEntries(auditFolder);
+ List<StoragePathInfo> auditFiles = new ArrayList<>();
+ List<StoragePathInfo> oldFiles = new ArrayList<>();
+
+ // Filter to get only .jsonl files
+ for (org.apache.hudi.storage.StoragePathInfo pathInfo : allFiles) {
+ if (pathInfo.isFile() &&
pathInfo.getPath().getName().endsWith(".jsonl")) {
+ auditFiles.add(pathInfo);
+ if (pathInfo.getModificationTime() < cutoffTime) {
+ oldFiles.add(pathInfo);
+ }
+ }
+ }
+
+ if (oldFiles.isEmpty()) {
+ String message = String.format("No audit files older than %d days
found.", ageDays);
+ return new CleanupResult(true, 0, 0, auditFiles.size(), message,
dryRun);
+ }
+
+ int fileCount = oldFiles.size();
+
+ if (dryRun) {
+ String message = String.format("Dry run: Would delete %d audit files
older than %d days.", fileCount, ageDays);
+ return new CleanupResult(true, 0, 0, fileCount, message, dryRun);
+ } else {
+ // Actually delete the files
+ int deletedCount = 0;
+ int failedCount = 0;
+
+ for (org.apache.hudi.storage.StoragePathInfo pathInfo : oldFiles) {
+ try {
+ HoodieCLI.storage.deleteFile(pathInfo.getPath());
+ deletedCount++;
+ } catch (Exception e) {
+ failedCount++;
+ LOG.warn("Failed to delete audit file: " + pathInfo.getPath(), e);
+ }
+ }
+
+ if (failedCount == 0) {
+ String message = String.format("Successfully deleted %d audit files
older than %d days.", deletedCount, ageDays);
+ return new CleanupResult(true, deletedCount, failedCount, fileCount,
message, dryRun);
+ } else {
+ String message = String.format("Deleted %d audit files, failed to
delete %d files.", deletedCount, failedCount);
+ return new CleanupResult(false, deletedCount, failedCount,
fileCount, message, dryRun);
+ }
+ }
+
+ } catch (Exception e) {
+ LOG.error("Error cleaning up audit locks", e);
+ String message = String.format("Failed to cleanup audit locks: %s",
e.getMessage());
+ return new CleanupResult(false, 0, 0, 0, message, dryRun);
+ }
+ }
+
+ /**
+ * Parses an audit file and extracts transaction window information.
+ */
+ private Option<TransactionWindow> parseAuditFile(StoragePathInfo pathInfo) {
+ String filename = pathInfo.getPath().getName();
+
+ try {
+ // Read file content using Hudi storage API
+ String content;
+ try (InputStream inputStream =
HoodieCLI.storage.open(pathInfo.getPath());
+ BufferedReader reader = new BufferedReader(new
InputStreamReader(inputStream))) {
+ StringBuilder sb = new StringBuilder();
+ String line;
+ while ((line = reader.readLine()) != null) {
+ sb.append(line).append("\n");
+ }
+ content = sb.toString();
+ }
+
+ // Parse JSONL content
+ String[] lines = content.split("\n");
+ List<JsonNode> jsonObjects = new ArrayList<>();
+ for (String line : lines) {
+ if (line.trim().isEmpty()) {
+ continue;
+ }
+ try {
+ jsonObjects.add(OBJECT_MAPPER.readTree(line));
+ } catch (Exception e) {
+ LOG.warn("Failed to parse JSON line in file " + filename + ": " +
line, e);
+ }
+ }
+
+ if (jsonObjects.isEmpty()) {
+ return Option.empty();
+ }
+
+ // Extract transaction metadata
+ JsonNode firstObject = jsonObjects.get(0);
+ String ownerId = firstObject.has("ownerId") ?
firstObject.get("ownerId").asText() : "unknown";
+ long transactionStartTime = firstObject.has("transactionStartTime")
+ ? firstObject.get("transactionStartTime").asLong() : 0L;
+
+ // Find first START timestamp
+ long startTimestamp = transactionStartTime; // default to transaction
start time
+ for (JsonNode obj : jsonObjects) {
+ if (obj.has("state") && "START".equals(obj.get("state").asText())) {
+ startTimestamp = obj.has("timestamp") ?
obj.get("timestamp").asLong() : transactionStartTime;
+ break;
+ }
+ }
Review Comment:
Should a `AudigLogEntry` class be created with `@JsonCreator` annotation for
serde with JSON so there is no such custom parsing logic?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/LockAuditingCommand.java:
##########
@@ -185,4 +272,346 @@ public String showLockAuditStatus() {
return String.format("Failed to check lock audit status: %s",
e.getMessage());
}
}
+
+ /**
+ * Validates the audit lock files for consistency and integrity.
+ * This command checks for issues such as corrupted files, invalid format,
+ * incorrect state transitions, and orphaned locks.
+ *
+ * @return Validation results including any issues found
+ */
+ @ShellMethod(key = "locks audit validate", value = "Validate audit lock
files for consistency and integrity")
+ public String validateAuditLocks() {
+
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ try {
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit folder found - nothing to validate";
+ }
Review Comment:
nit: catch `FileNotFoundException` from `listDirectEntries` to indicate that
the folder does not exist and avoid `storage.exists` check?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/LockAuditingCommand.java:
##########
@@ -185,4 +272,346 @@ public String showLockAuditStatus() {
return String.format("Failed to check lock audit status: %s",
e.getMessage());
}
}
+
+ /**
+ * Validates the audit lock files for consistency and integrity.
+ * This command checks for issues such as corrupted files, invalid format,
+ * incorrect state transitions, and orphaned locks.
+ *
+ * @return Validation results including any issues found
+ */
+ @ShellMethod(key = "locks audit validate", value = "Validate audit lock
files for consistency and integrity")
+ public String validateAuditLocks() {
+
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ try {
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit folder found - nothing to validate";
+ }
+
+ // Get all audit files
+ List<StoragePathInfo> allFiles =
HoodieCLI.storage.listDirectEntries(auditFolder);
+ List<StoragePathInfo> auditFiles = new ArrayList<>();
+ for (StoragePathInfo pathInfo : allFiles) {
+ if (pathInfo.isFile() &&
pathInfo.getPath().getName().endsWith(".jsonl")) {
+ auditFiles.add(pathInfo);
+ }
+ }
+
+ if (auditFiles.isEmpty()) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit files found - nothing to validate";
+ }
+
+ // Parse all audit files into transaction windows
+ List<TransactionWindow> windows = new ArrayList<>();
+ for (StoragePathInfo pathInfo : auditFiles) {
+ Option<TransactionWindow> window = parseAuditFile(pathInfo);
+ if (window.isPresent()) {
+ windows.add(window.get());
+ }
+ }
+
+ if (windows.isEmpty()) {
+ return String.format("Validation Result: FAILED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: %d\n"
+ + "Details: Failed to parse any audit files", auditFiles.size());
+ }
+
+ // Validate transactions
+ ValidationResults validationResults =
validateTransactionWindows(windows);
+
+ // Generate result
+ int totalIssues = validationResults.errors.size() +
validationResults.warnings.size();
+ String result;
+ String details;
+
+ if (totalIssues == 0) {
+ result = "PASSED";
+ details = "All audit lock transactions validated successfully";
+ } else {
+ result = validationResults.errors.isEmpty() ? "WARNING" : "FAILED";
+ List<String> allIssues = new ArrayList<>();
+ allIssues.addAll(validationResults.errors);
+ allIssues.addAll(validationResults.warnings);
+ details = String.join(", ", allIssues);
+ }
+
+ return String.format("Validation Result: %s\n"
+ + "Transactions Validated: %d\n"
+ + "Issues Found: %d\n"
+ + "Details: %s", result, windows.size(), totalIssues, details);
+
+ } catch (Exception e) {
+ LOG.error("Error validating audit locks", e);
+ return String.format("Validation Result: ERROR\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: -1\n"
+ + "Details: Validation failed: %s", e.getMessage());
+ }
+ }
+
+ /**
+ * Cleans up old audit lock files based on age threshold.
+ * This command removes audit files that are older than the specified number
of days.
+ *
+ * @param dryRun Whether to perform a dry run (preview changes without
deletion)
+ * @param ageDays Number of days to keep audit files (default 7)
+ * @return Status message indicating files cleaned or to be cleaned
+ */
+ @ShellMethod(key = "locks audit cleanup", value = "Clean up old audit lock
files")
+ public String cleanupAuditLocks(
+ @ShellOption(value = {"--dryRun"}, defaultValue = "false",
+ help = "Preview changes without actually deleting files") final
boolean dryRun,
+ @ShellOption(value = {"--ageDays"}, defaultValue = "7",
+ help = "Delete audit files older than this many days") final String
ageDaysStr) {
+
+ try {
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ // Parse ageDays manually to handle validation properly
+ int ageDays;
+ try {
+ ageDays = Integer.parseInt(ageDaysStr);
+ } catch (NumberFormatException e) {
+ return "Error: ageDays must be a value greater than 0.";
+ }
+
+ if (ageDays < 0) {
+ return "Error: ageDays must be non-negative (>= 0).";
+ }
Review Comment:
nit: remove this since it's already checked in `performAuditCleanup`?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/LockAuditingCommand.java:
##########
@@ -44,6 +51,79 @@ public class LockAuditingCommand {
private static final Logger LOG =
LoggerFactory.getLogger(LockAuditingCommand.class);
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ /**
+ * Represents a transaction window with start time, end time, and metadata.
+ */
+ static class TransactionWindow {
+ final String ownerId;
+ final long transactionStartTime;
+ final long startTimestamp;
+ final Option<Long> endTimestamp;
+ final Option<Long> lastExpirationTime;
Review Comment:
What is the difference between `transactionStartTime` and `startTimestamp`?
Name the latter and `endTimestamp`, `lastExpirationTime` better so it's clear
from the naming. Also, use either `Time` or `Timestamp` for consistency.
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/LockAuditingCommand.java:
##########
@@ -185,4 +272,346 @@ public String showLockAuditStatus() {
return String.format("Failed to check lock audit status: %s",
e.getMessage());
}
}
+
+ /**
+ * Validates the audit lock files for consistency and integrity.
+ * This command checks for issues such as corrupted files, invalid format,
+ * incorrect state transitions, and orphaned locks.
+ *
+ * @return Validation results including any issues found
+ */
+ @ShellMethod(key = "locks audit validate", value = "Validate audit lock
files for consistency and integrity")
+ public String validateAuditLocks() {
+
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ try {
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit folder found - nothing to validate";
+ }
+
+ // Get all audit files
+ List<StoragePathInfo> allFiles =
HoodieCLI.storage.listDirectEntries(auditFolder);
+ List<StoragePathInfo> auditFiles = new ArrayList<>();
+ for (StoragePathInfo pathInfo : allFiles) {
+ if (pathInfo.isFile() &&
pathInfo.getPath().getName().endsWith(".jsonl")) {
+ auditFiles.add(pathInfo);
+ }
+ }
+
+ if (auditFiles.isEmpty()) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit files found - nothing to validate";
+ }
+
+ // Parse all audit files into transaction windows
+ List<TransactionWindow> windows = new ArrayList<>();
+ for (StoragePathInfo pathInfo : auditFiles) {
+ Option<TransactionWindow> window = parseAuditFile(pathInfo);
+ if (window.isPresent()) {
+ windows.add(window.get());
+ }
+ }
+
+ if (windows.isEmpty()) {
+ return String.format("Validation Result: FAILED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: %d\n"
+ + "Details: Failed to parse any audit files", auditFiles.size());
+ }
Review Comment:
Even if `windows` is non-empty, there can be exception caught within
`parseAuditFile` which is swallowed?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/LockAuditingCommand.java:
##########
@@ -185,4 +272,346 @@ public String showLockAuditStatus() {
return String.format("Failed to check lock audit status: %s",
e.getMessage());
}
}
+
+ /**
+ * Validates the audit lock files for consistency and integrity.
+ * This command checks for issues such as corrupted files, invalid format,
+ * incorrect state transitions, and orphaned locks.
+ *
+ * @return Validation results including any issues found
+ */
+ @ShellMethod(key = "locks audit validate", value = "Validate audit lock
files for consistency and integrity")
+ public String validateAuditLocks() {
+
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ try {
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit folder found - nothing to validate";
+ }
+
+ // Get all audit files
+ List<StoragePathInfo> allFiles =
HoodieCLI.storage.listDirectEntries(auditFolder);
+ List<StoragePathInfo> auditFiles = new ArrayList<>();
+ for (StoragePathInfo pathInfo : allFiles) {
+ if (pathInfo.isFile() &&
pathInfo.getPath().getName().endsWith(".jsonl")) {
+ auditFiles.add(pathInfo);
+ }
+ }
+
+ if (auditFiles.isEmpty()) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit files found - nothing to validate";
+ }
+
+ // Parse all audit files into transaction windows
+ List<TransactionWindow> windows = new ArrayList<>();
+ for (StoragePathInfo pathInfo : auditFiles) {
+ Option<TransactionWindow> window = parseAuditFile(pathInfo);
+ if (window.isPresent()) {
+ windows.add(window.get());
+ }
+ }
+
+ if (windows.isEmpty()) {
+ return String.format("Validation Result: FAILED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: %d\n"
+ + "Details: Failed to parse any audit files", auditFiles.size());
+ }
+
+ // Validate transactions
+ ValidationResults validationResults =
validateTransactionWindows(windows);
+
+ // Generate result
+ int totalIssues = validationResults.errors.size() +
validationResults.warnings.size();
+ String result;
+ String details;
+
+ if (totalIssues == 0) {
+ result = "PASSED";
+ details = "All audit lock transactions validated successfully";
+ } else {
+ result = validationResults.errors.isEmpty() ? "WARNING" : "FAILED";
+ List<String> allIssues = new ArrayList<>();
+ allIssues.addAll(validationResults.errors);
+ allIssues.addAll(validationResults.warnings);
+ details = String.join(", ", allIssues);
+ }
+
+ return String.format("Validation Result: %s\n"
+ + "Transactions Validated: %d\n"
+ + "Issues Found: %d\n"
+ + "Details: %s", result, windows.size(), totalIssues, details);
+
+ } catch (Exception e) {
+ LOG.error("Error validating audit locks", e);
+ return String.format("Validation Result: ERROR\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: -1\n"
+ + "Details: Validation failed: %s", e.getMessage());
+ }
+ }
+
+ /**
+ * Cleans up old audit lock files based on age threshold.
+ * This command removes audit files that are older than the specified number
of days.
+ *
+ * @param dryRun Whether to perform a dry run (preview changes without
deletion)
+ * @param ageDays Number of days to keep audit files (default 7)
+ * @return Status message indicating files cleaned or to be cleaned
+ */
+ @ShellMethod(key = "locks audit cleanup", value = "Clean up old audit lock
files")
+ public String cleanupAuditLocks(
+ @ShellOption(value = {"--dryRun"}, defaultValue = "false",
+ help = "Preview changes without actually deleting files") final
boolean dryRun,
+ @ShellOption(value = {"--ageDays"}, defaultValue = "7",
+ help = "Delete audit files older than this many days") final String
ageDaysStr) {
+
+ try {
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ // Parse ageDays manually to handle validation properly
+ int ageDays;
+ try {
+ ageDays = Integer.parseInt(ageDaysStr);
+ } catch (NumberFormatException e) {
+ return "Error: ageDays must be a value greater than 0.";
+ }
+
+ if (ageDays < 0) {
+ return "Error: ageDays must be non-negative (>= 0).";
+ }
+
+ return performAuditCleanup(dryRun, ageDays).toString();
+ } catch (Exception e) {
+ LOG.error("Error during audit cleanup", e);
+ return String.format("Error during cleanup: %s", e.getMessage() != null
? e.getMessage() : e.getClass().getSimpleName());
+ }
+ }
+
+ /**
+ * Internal method to perform audit cleanup. Used by both the CLI command
and disable method.
+ *
+ * @param dryRun Whether to perform a dry run (preview changes without
deletion)
+ * @param ageDays Number of days to keep audit files (0 means delete all)
+ * @return CleanupResult containing cleanup operation details
+ */
+ private CleanupResult performAuditCleanup(boolean dryRun, int ageDays) {
+ if (ageDays < 0) {
+ String message = "Error: ageDays must be non-negative (>= 0).";
+ return new CleanupResult(false, 0, 0, 0, message, dryRun);
+ }
+
+ try {
+ if (HoodieCLI.storage == null) {
+ String message = "Storage not initialized.";
+ return new CleanupResult(false, 0, 0, 0, message, dryRun);
+ }
+
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ String message = "No audit folder found - nothing to cleanup.";
+ return new CleanupResult(true, 0, 0, 0, message, dryRun);
+ }
+
+ // Calculate cutoff timestamp (ageDays ago)
+ long cutoffTime = System.currentTimeMillis() - (ageDays * 24L * 60L *
60L * 1000L);
+
+ // List all files in audit folder and filter by modification time
+ List<StoragePathInfo> allFiles =
HoodieCLI.storage.listDirectEntries(auditFolder);
+ List<StoragePathInfo> auditFiles = new ArrayList<>();
+ List<StoragePathInfo> oldFiles = new ArrayList<>();
+
+ // Filter to get only .jsonl files
+ for (org.apache.hudi.storage.StoragePathInfo pathInfo : allFiles) {
Review Comment:
```suggestion
for (StoragePathInfo pathInfo : allFiles) {
```
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/LockAuditingCommand.java:
##########
@@ -185,4 +272,346 @@ public String showLockAuditStatus() {
return String.format("Failed to check lock audit status: %s",
e.getMessage());
}
}
+
+ /**
+ * Validates the audit lock files for consistency and integrity.
+ * This command checks for issues such as corrupted files, invalid format,
+ * incorrect state transitions, and orphaned locks.
+ *
+ * @return Validation results including any issues found
+ */
+ @ShellMethod(key = "locks audit validate", value = "Validate audit lock
files for consistency and integrity")
+ public String validateAuditLocks() {
+
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ try {
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit folder found - nothing to validate";
+ }
+
+ // Get all audit files
+ List<StoragePathInfo> allFiles =
HoodieCLI.storage.listDirectEntries(auditFolder);
+ List<StoragePathInfo> auditFiles = new ArrayList<>();
+ for (StoragePathInfo pathInfo : allFiles) {
+ if (pathInfo.isFile() &&
pathInfo.getPath().getName().endsWith(".jsonl")) {
+ auditFiles.add(pathInfo);
+ }
+ }
+
+ if (auditFiles.isEmpty()) {
+ return "Validation Result: PASSED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: 0\n"
+ + "Details: No audit files found - nothing to validate";
+ }
+
+ // Parse all audit files into transaction windows
+ List<TransactionWindow> windows = new ArrayList<>();
+ for (StoragePathInfo pathInfo : auditFiles) {
+ Option<TransactionWindow> window = parseAuditFile(pathInfo);
+ if (window.isPresent()) {
+ windows.add(window.get());
+ }
+ }
+
+ if (windows.isEmpty()) {
+ return String.format("Validation Result: FAILED\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: %d\n"
+ + "Details: Failed to parse any audit files", auditFiles.size());
+ }
+
+ // Validate transactions
+ ValidationResults validationResults =
validateTransactionWindows(windows);
+
+ // Generate result
+ int totalIssues = validationResults.errors.size() +
validationResults.warnings.size();
+ String result;
+ String details;
+
+ if (totalIssues == 0) {
+ result = "PASSED";
+ details = "All audit lock transactions validated successfully";
+ } else {
+ result = validationResults.errors.isEmpty() ? "WARNING" : "FAILED";
+ List<String> allIssues = new ArrayList<>();
+ allIssues.addAll(validationResults.errors);
+ allIssues.addAll(validationResults.warnings);
+ details = String.join(", ", allIssues);
+ }
+
+ return String.format("Validation Result: %s\n"
+ + "Transactions Validated: %d\n"
+ + "Issues Found: %d\n"
+ + "Details: %s", result, windows.size(), totalIssues, details);
+
+ } catch (Exception e) {
+ LOG.error("Error validating audit locks", e);
+ return String.format("Validation Result: ERROR\n"
+ + "Transactions Validated: 0\n"
+ + "Issues Found: -1\n"
+ + "Details: Validation failed: %s", e.getMessage());
+ }
+ }
+
+ /**
+ * Cleans up old audit lock files based on age threshold.
+ * This command removes audit files that are older than the specified number
of days.
+ *
+ * @param dryRun Whether to perform a dry run (preview changes without
deletion)
+ * @param ageDays Number of days to keep audit files (default 7)
+ * @return Status message indicating files cleaned or to be cleaned
+ */
+ @ShellMethod(key = "locks audit cleanup", value = "Clean up old audit lock
files")
+ public String cleanupAuditLocks(
+ @ShellOption(value = {"--dryRun"}, defaultValue = "false",
+ help = "Preview changes without actually deleting files") final
boolean dryRun,
+ @ShellOption(value = {"--ageDays"}, defaultValue = "7",
+ help = "Delete audit files older than this many days") final String
ageDaysStr) {
+
+ try {
+ if (HoodieCLI.basePath == null) {
+ return "No Hudi table loaded. Please connect to a table first.";
+ }
+
+ // Parse ageDays manually to handle validation properly
+ int ageDays;
+ try {
+ ageDays = Integer.parseInt(ageDaysStr);
+ } catch (NumberFormatException e) {
+ return "Error: ageDays must be a value greater than 0.";
+ }
+
+ if (ageDays < 0) {
+ return "Error: ageDays must be non-negative (>= 0).";
+ }
+
+ return performAuditCleanup(dryRun, ageDays).toString();
+ } catch (Exception e) {
+ LOG.error("Error during audit cleanup", e);
+ return String.format("Error during cleanup: %s", e.getMessage() != null
? e.getMessage() : e.getClass().getSimpleName());
+ }
+ }
+
+ /**
+ * Internal method to perform audit cleanup. Used by both the CLI command
and disable method.
+ *
+ * @param dryRun Whether to perform a dry run (preview changes without
deletion)
+ * @param ageDays Number of days to keep audit files (0 means delete all)
+ * @return CleanupResult containing cleanup operation details
+ */
+ private CleanupResult performAuditCleanup(boolean dryRun, int ageDays) {
+ if (ageDays < 0) {
+ String message = "Error: ageDays must be non-negative (>= 0).";
+ return new CleanupResult(false, 0, 0, 0, message, dryRun);
+ }
+
+ try {
+ if (HoodieCLI.storage == null) {
+ String message = "Storage not initialized.";
+ return new CleanupResult(false, 0, 0, 0, message, dryRun);
+ }
+
+ String auditFolderPath =
StorageLockProviderAuditService.getAuditFolderPath(HoodieCLI.basePath);
+ StoragePath auditFolder = new StoragePath(auditFolderPath);
+
+ // Check if audit folder exists
+ if (!HoodieCLI.storage.exists(auditFolder)) {
+ String message = "No audit folder found - nothing to cleanup.";
+ return new CleanupResult(true, 0, 0, 0, message, dryRun);
+ }
Review Comment:
Similarly, avoid `exists` call and catch `FileNotFoundException from
`storage.listDirectEntries`?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
