[
https://issues.apache.org/jira/browse/HADOOP-19445?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17932231#comment-17932231
]
ASF GitHub Bot commented on HADOOP-19445:
-----------------------------------------
bhattmanish98 commented on code in PR #7386:
URL: https://github.com/apache/hadoop/pull/7386#discussion_r1979065621
##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRename.java:
##########
@@ -1655,6 +1730,827 @@ public void
testRenameSrcDirDeleteEmitDeletionCountInClientRequestId()
fs.rename(new Path(dirPathStr), new Path("/dst/"));
}
+ /**
+ * Helper method to configure the AzureBlobFileSystem and rename directories.
+ *
+ * @param currentFs The current AzureBlobFileSystem to use for renaming.
+ * @param producerQueueSize Maximum size of the producer queue.
+ * @param consumerMaxLag Maximum lag allowed for the consumer.
+ * @param maxThread Maximum threads for the rename operation.
+ * @param src The source path of the directory to rename.
+ * @param dst The destination path of the renamed directory.
+ * @throws IOException If an I/O error occurs during the operation.
+ */
+ private void renameDir(AzureBlobFileSystem currentFs, String
producerQueueSize,
+ String consumerMaxLag, String maxThread, Path src, Path dst)
+ throws IOException {
+ Configuration config = createConfig(producerQueueSize, consumerMaxLag,
maxThread);
+ try (AzureBlobFileSystem fs = (AzureBlobFileSystem)
FileSystem.newInstance(currentFs.getUri(), config)) {
+ fs.rename(src, dst);
+ validateRename(fs, src, dst, false, true, false);
+ }
+ }
+
+ /**
+ * Helper method to create the configuration for the AzureBlobFileSystem.
+ *
+ * @param producerQueueSize Maximum size of the producer queue.
+ * @param consumerMaxLag Maximum lag allowed for the consumer.
+ * @param maxThread Maximum threads for the rename operation.
+ * @return The configuration object.
+ */
+ private Configuration createConfig(String producerQueueSize, String
consumerMaxLag, String maxThread) {
+ Configuration config = new Configuration(this.getRawConfiguration());
+ config.set(FS_AZURE_PRODUCER_QUEUE_MAX_SIZE, producerQueueSize);
+ config.set(FS_AZURE_CONSUMER_MAX_LAG, consumerMaxLag);
+ config.set(FS_AZURE_BLOB_DIR_RENAME_MAX_THREAD, maxThread);
+ return config;
+ }
+
+ /**
+ * Helper method to validate that the rename was successful and that the
destination exists.
+ *
+ * @param fs The AzureBlobFileSystem instance to check the existence on.
+ * @param dst The destination path.
+ * @param src The source path.
+ * @throws IOException If an I/O error occurs during the validation.
+ */
+ private void validateRename(AzureBlobFileSystem fs, Path src, Path dst,
+ boolean isSrcExist, boolean isDstExist, boolean isJsonExist)
+ throws IOException {
+ Assertions.assertThat(fs.exists(dst))
+ .describedAs("Renamed Destination directory should exist.")
+ .isEqualTo(isDstExist);
+ Assertions.assertThat(fs.exists(new Path(src.getParent(), src.getName() +
SUFFIX)))
+ .describedAs("Renamed Pending Json file should exist.")
+ .isEqualTo(isJsonExist);
+ Assertions.assertThat(fs.exists(src))
+ .describedAs("Renamed Destination directory should exist.")
+ .isEqualTo(isSrcExist);
+ }
+
+ /**
+ * Test the renaming of a directory with different parallelism
configurations.
+ */
+ @Test
+ public void testRenameDirWithDifferentParallelism() throws Exception {
+ try (AzureBlobFileSystem currentFs = getFileSystem()) {
+ assumeBlobServiceType();
+ Path src = new Path("/hbase/A1/A2");
+ Path dst = new Path("/hbase/A1/A3");
+
+ // Create sample files in the source directory
+ createFiles(currentFs, src, TOTAL_FILES);
+
+ // Test renaming with different configurations
+ renameDir(currentFs, "10", "5", "2", src, dst);
+ renameDir(currentFs, "100", "5", "2", dst, src);
+
+ String errorMessage = intercept(PathIOException.class,
+ () -> renameDir(currentFs, "50", "50", "5", src, dst))
+ .getMessage();
+
+ // Validate error message for invalid configuration
+ Assertions.assertThat(errorMessage)
+ .describedAs("maxConsumptionLag should be lesser than maxSize")
+ .contains(
+ "Invalid configuration value detected for
\"fs.azure.blob.dir.list.consumer.max.lag\". "
+ + "maxConsumptionLag should be lesser than maxSize");
+ }
+ }
+
+ /**
+ * Helper method to create files in the given directory.
+ *
+ * @param fs The AzureBlobFileSystem instance to use for file creation.
+ * @param src The source path (directory).
+ * @param numFiles The number of files to create.
+ * @throws ExecutionException, InterruptedException If an error occurs
during file creation.
+ */
+ private void createFiles(AzureBlobFileSystem fs, Path src, int numFiles)
+ throws ExecutionException, InterruptedException {
+ ExecutorService executorService =
Executors.newFixedThreadPool(TOTAL_THREADS_IN_POOL);
+ List<Future> futures = new ArrayList<>();
+ for (int i = 0; i < numFiles; i++) {
+ final int iter = i;
+ Future future = executorService.submit(() ->
+ fs.create(new Path(src, "file" + iter + ".txt")));
+ futures.add(future);
+ }
+ for (Future future : futures) {
+ future.get();
+ }
+ executorService.shutdown();
+ }
+
+ /**
+ * Tests renaming a directory with a failure during the copy operation.
+ * Simulates an error when copying on the 6th call.
+ */
+ @Test
+ public void testRenameCopyFailureInBetween() throws Exception {
+ try (AzureBlobFileSystem fs = Mockito.spy(this.getFileSystem(
+ createConfig("5", "3", "2")))) {
+ assumeBlobServiceType();
+ AbfsBlobClient client = (AbfsBlobClient) addSpyHooksOnClient(fs);
+ fs.getAbfsStore().setClient(client);
+ Path src = new Path("/hbase/A1/A2");
+ Path dst = new Path("/hbase/A1/A3");
+
+ // Create sample files in the source directory
+ createFiles(fs, src, TOTAL_FILES);
+
+ // Track the number of copy operations
+ AtomicInteger copyCall = new AtomicInteger(0);
+ Mockito.doAnswer(copyRequest -> {
+ if (copyCall.get() == FAILED_CALL) {
+ throw new AbfsRestOperationException(
+ BLOB_ALREADY_EXISTS.getStatusCode(),
+ BLOB_ALREADY_EXISTS.getErrorCode(),
+ BLOB_ALREADY_EXISTS.getErrorMessage(),
+ new Exception());
+ }
+ copyCall.incrementAndGet();
+ return copyRequest.callRealMethod();
+ }).when(client).copyBlob(Mockito.any(Path.class),
+ Mockito.any(Path.class), Mockito.nullable(String.class),
+ Mockito.any(TracingContext.class));
+
+ fs.rename(src, dst);
+ // Validate copy operation count
+ Assertions.assertThat(copyCall.get())
+ .describedAs("Copy operation count should be less than 10.")
+ .isLessThan(TOTAL_FILES);
+
+ // Validate that rename redo operation was triggered
+ copyCall.set(0);
+
+ // Assertions to validate renamed destination and source
+ validateRename(fs, src, dst, false, true, true);
+
+ Assertions.assertThat(copyCall.get())
+ .describedAs("Copy operation count should be greater than 0.")
+ .isGreaterThan(0);
+
+ // Validate final state of destination and source
+ validateRename(fs, src, dst, false, true, false);
+ }
+ }
+
+ /**
+ * Tests renaming a directory with a failure during the delete operation.
+ * Simulates an error on the 6th delete operation and verifies the behavior.
+ */
+ @Test
+ public void testRenameDeleteFailureInBetween() throws Exception {
+ try (AzureBlobFileSystem fs = Mockito.spy(this.getFileSystem(
+ createConfig("5", "3", "2")))) {
+ assumeBlobServiceType();
+ AbfsBlobClient client = (AbfsBlobClient) addSpyHooksOnClient(fs);
+ fs.getAbfsStore().setClient(client);
+ Path src = new Path("/hbase/A1/A2");
+ Path dst = new Path("/hbase/A1/A3");
+
+ // Create sample files in the source directory
+ createFiles(fs, src, TOTAL_FILES);
+
+ // Track the number of delete operations
+ AtomicInteger deleteCall = new AtomicInteger(0);
+ Mockito.doAnswer(deleteRequest -> {
+ if (deleteCall.get() == FAILED_CALL) {
+ throw new AbfsRestOperationException(
+ BLOB_PATH_NOT_FOUND.getStatusCode(),
+ BLOB_PATH_NOT_FOUND.getErrorCode(),
+ BLOB_PATH_NOT_FOUND.getErrorMessage(),
+ new Exception());
+ }
+ deleteCall.incrementAndGet();
+ return deleteRequest.callRealMethod();
+ }).when(client).deleteBlobPath(Mockito.any(Path.class),
+ Mockito.anyString(), Mockito.any(TracingContext.class));
+
+ fs.rename(src, dst);
+
+ // Validate delete operation count
+ Assertions.assertThat(deleteCall.get())
+ .describedAs("Delete operation count should be less than 10.")
+ .isLessThan(TOTAL_FILES);
+
+ // Validate that delete redo operation was triggered
+ deleteCall.set(0);
+ // Assertions to validate renamed destination and source
+ validateRename(fs, src, dst, false, true, true);
+
+ Assertions.assertThat(deleteCall.get())
+ .describedAs("Delete operation count should be greater than 0.")
+ .isGreaterThan(0);
+
+ // Validate final state of destination and source
+ // Validate that delete redo operation was triggered
+ validateRename(fs, src, dst, false, true, false);
+ }
+ }
+
+ /**
+ * Tests renaming a file or directory when the destination path contains
+ * a colon (":"). The test ensures that:
+ * - The source directory exists before the rename.
+ * - The file is successfully renamed to the destination path.
+ * - The old source directory no longer exists after the rename.
+ * - The new destination directory exists after the rename.
+ *
+ * @throws Exception if an error occurs during file system operations
+ */
+ @Test
+ public void testRenameWhenDestinationPathContainsColon() throws Exception {
+ AzureBlobFileSystem fs = getFileSystem();
+ fs.setWorkingDirectory(new Path(ROOT_PATH));
+ String fileName = "file";
+ Path src = new Path("/test1/");
+ Path dst = new Path("/test1:/");
+
+ // Create the file
+ fs.create(new Path(src, fileName));
+
+ // Perform the rename operation and validate the results
+ performRenameAndValidate(fs, src, dst, fileName);
+ }
+
+ /**
+ * Performs the rename operation and validates the existence of the
directories and files.
+ *
+ * @param fs the AzureBlobFileSystem instance
+ * @param src the source path to be renamed
+ * @param dst the destination path for the rename
+ * @param fileName the name of the file to be renamed
+ */
+ private void performRenameAndValidate(AzureBlobFileSystem fs, Path src, Path
dst, String fileName)
+ throws IOException {
+ // Assert the source directory exists
+ Assertions.assertThat(fs.exists(src))
+ .describedAs("Old directory should exist before rename")
+ .isTrue();
+
+ // Perform rename
+ fs.rename(src, dst);
+
+ // Assert the destination directory and file exist after rename
+ Assertions.assertThat(fs.exists(new Path(dst, fileName)))
+ .describedAs("Rename should be successful")
+ .isTrue();
+
+ // Assert the source directory no longer exists
+ Assertions.assertThat(fs.exists(src))
+ .describedAs("Old directory should not exist")
+ .isFalse();
+
+ // Assert the new destination directory exists
+ Assertions.assertThat(fs.exists(dst))
+ .describedAs("New directory should exist")
+ .isTrue();
+ }
+
+ /**
+ * Tests the behavior of the atomic rename key for the root folder
+ * in Azure Blob File System. The test verifies that the atomic rename key
+ * returns false for the root folder path.
+ *
+ * @throws Exception if an error occurs during the atomic rename key check
+ */
+ @Test
+ public void testGetAtomicRenameKeyForRootFolder() throws Exception {
+ AzureBlobFileSystem fs = getFileSystem();
+ assumeBlobServiceType();
+ AbfsBlobClient abfsBlobClient = (AbfsBlobClient) fs.getAbfsClient();
+ Assertions.assertThat(abfsBlobClient.isAtomicRenameKey("/hbase"))
+ .describedAs("Atomic rename key should return false for Root folder")
+ .isFalse();
+ }
+
+ /**
+ * Tests the behavior of the atomic rename key for non-root folders
+ * in Azure Blob File System. The test verifies that the atomic rename key
+ * works for specific folders as defined in the configuration.
+ * It checks the atomic rename key for various paths,
+ * ensuring it returns true for matching paths and false for others.
+ *
+ * @throws Exception if an error occurs during the atomic rename key check
+ */
+ @Test
+ public void testGetAtomicRenameKeyForNonRootFolder() throws Exception {
+ final AzureBlobFileSystem currentFs = getFileSystem();
+ Configuration config = new Configuration(this.getRawConfiguration());
+ config.set(FS_AZURE_ATOMIC_RENAME_KEY, "/hbase,/a,/b");
+
+ final AzureBlobFileSystem fs = (AzureBlobFileSystem)
FileSystem.newInstance(currentFs.getUri(), config);
+ assumeBlobServiceType();
+ AbfsBlobClient abfsBlobClient = (AbfsBlobClient) fs.getAbfsClient();
+
+ // Test for various paths
+ validateAtomicRenameKey(abfsBlobClient, "/hbase1/test", false);
+ validateAtomicRenameKey(abfsBlobClient, "/hbase/test", true);
+ validateAtomicRenameKey(abfsBlobClient, "/a/b/c", true);
+ validateAtomicRenameKey(abfsBlobClient, "/test/a", false);
+ }
+
+ /**
+ * Validates the atomic rename key for a specific path.
+ *
+ * @param abfsBlobClient the AbfsBlobClient instance
+ * @param path the path to check for atomic rename key
+ * @param expected the expected value (true or false)
+ */
+ private void validateAtomicRenameKey(AbfsBlobClient abfsBlobClient, String
path, boolean expected) {
+ Assertions.assertThat(abfsBlobClient.isAtomicRenameKey(path))
+ .describedAs("Atomic rename key check for path: " + path)
+ .isEqualTo(expected);
+ }
+
+ /**
+ * Helper method to create a json file.
+ * @param path parent path
+ * @param renameJson rename json path
+ * @return file system
+ * @throws IOException in case of failure
+ */
+ public AzureBlobFileSystem createJsonFile(Path path, Path renameJson) throws
IOException {
+ final AzureBlobFileSystem fs = Mockito.spy(this.getFileSystem());
+ assumeBlobServiceType();
+ AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
+ Mockito.doReturn(store).when(fs).getAbfsStore();
+ AbfsClient client = Mockito.spy(store.getClient());
+ Mockito.doReturn(client).when(store).getClient();
+
+ fs.setWorkingDirectory(new Path(ROOT_PATH));
+ fs.create(new Path(path, "file.txt"));
+
+ AzureBlobFileSystemStore.VersionedFileStatus fileStatus =
(AzureBlobFileSystemStore.VersionedFileStatus) fs.getFileStatus(path);
+
+ new RenameAtomicity(path, new Path("/hbase/test4"), renameJson,
getTestTracingContext(fs, true), fileStatus.getEtag(), client)
+ .preRename();
+
+ Assertions.assertThat(fs.exists(renameJson))
+ .describedAs("Rename Pending Json file should exist.")
+ .isTrue();
+
+ return fs;
+ }
+
+ /**
+ * Test case to verify crash recovery with a single child folder.
+ *
+ * This test simulates a scenario where a pending rename JSON file exists
for a single child folder
+ * under the parent directory. It ensures that when listing the files in the
parent directory,
+ * only the child folder (with the pending rename JSON file) is returned,
and no additional files are listed.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void listCrashRecoveryWithSingleChildFolder() throws Exception {
+ AzureBlobFileSystem fs = null;
+ try {
+ Path path = new Path("/hbase/A1/A2");
+ Path renameJson = new Path(path.getParent(), path.getName() + SUFFIX);
+ fs = createJsonFile(path, renameJson);
+
+ FileStatus[] fileStatuses = fs.listStatus(new Path("/hbase/A1"));
+
+ Assertions.assertThat(fileStatuses.length)
+ .describedAs("List should return 1 file")
+ .isEqualTo(1);
+ } finally {
+ if (fs != null) {
+ fs.close();
+ }
+ }
+ }
+
+ /**
+ * Test case to verify crash recovery with multiple child folders.
+ *
+ * This test simulates a scenario where a pending rename JSON file exists,
and multiple files are
+ * created in the parent directory. It ensures that when listing the files
in the parent directory,
+ * the correct number of files is returned, including the pending rename
JSON file.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void listCrashRecoveryWithMultipleChildFolder() throws Exception {
+ AzureBlobFileSystem fs = null;
+ try {
+ Path path = new Path("/hbase/A1/A2");
+ Path renameJson = new Path(path.getParent(), path.getName() + SUFFIX);
+ fs = createJsonFile(path, renameJson);
+
+ fs.create(new Path("/hbase/A1/file1.txt"));
+ fs.create(new Path("/hbase/A1/file2.txt"));
+
+ FileStatus[] fileStatuses = fs.listStatus(new Path("/hbase/A1"));
+
+ Assertions.assertThat(fileStatuses.length)
+ .describedAs("List should return 3 files")
+ .isEqualTo(3);
+ } finally {
+ if (fs != null) {
+ fs.close();
+ }
+ }
+ }
+
+ /**
+ * Test case to verify crash recovery with a pending rename JSON file.
+ *
+ * This test simulates a scenario where a pending rename JSON file exists in
the parent directory,
+ * and it ensures that after the deletion of the target directory and
creation of new files,
+ * the listing operation correctly returns the remaining files without
considering the pending rename.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void listCrashRecoveryWithPendingJsonFile() throws Exception {
+ AzureBlobFileSystem fs = null;
+ try {
+ Path path = new Path("/hbase/A1/A2");
+ Path renameJson = new Path(path.getParent(), path.getName() + SUFFIX);
+ fs = createJsonFile(path, renameJson);
+
+ fs.delete(path, true);
+ fs.create(new Path("/hbase/A1/file1.txt"));
+ fs.create(new Path("/hbase/A1/file2.txt"));
+
+ FileStatus[] fileStatuses = fs.listStatus(path.getParent());
+
+ Assertions.assertThat(fileStatuses.length)
+ .describedAs("List should return 2 files")
+ .isEqualTo(2);
+ } finally {
+ if (fs != null) {
+ fs.close();
+ }
+ }
+ }
+
+ /**
+ * Test case to verify crash recovery when no pending rename JSON file
exists.
+ *
+ * This test simulates a scenario where there is no pending rename JSON file
in the directory.
+ * It ensures that the listing operation correctly returns all files in the
parent directory, including
+ * those created after the rename JSON file is deleted.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void listCrashRecoveryWithoutAnyPendingJsonFile() throws Exception {
+ AzureBlobFileSystem fs = null;
+ try {
+ Path path = new Path("/hbase/A1/A2");
+ Path renameJson = new Path(path.getParent(), path.getName() + SUFFIX);
+ fs = createJsonFile(path, renameJson);
+
+ fs.delete(renameJson, true);
+ fs.create(new Path("/hbase/A1/file1.txt"));
+ fs.create(new Path("/hbase/A1/file2.txt"));
+
+ FileStatus[] fileStatuses = fs.listStatus(path.getParent());
+
+ Assertions.assertThat(fileStatuses.length)
+ .describedAs("List should return 3 files")
+ .isEqualTo(3);
+ } finally {
+ if (fs != null) {
+ fs.close();
+ }
+ }
+ }
+
+ /**
+ * Test case to verify crash recovery when a pending rename JSON directory
exists.
+ *
+ * This test simulates a scenario where a pending rename JSON directory
exists, ensuring that the
+ * listing operation correctly returns all files in the parent directory
without triggering a redo
+ * rename operation. It also checks that the directory with the suffix
"-RenamePending.json" exists.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void listCrashRecoveryWithPendingJsonDir() throws Exception {
+ try (AzureBlobFileSystem fs = Mockito.spy(this.getFileSystem())) {
+ assumeBlobServiceType();
+ AbfsBlobClient client = (AbfsBlobClient) addSpyHooksOnClient(fs);
+
+ Path path = new Path("/hbase/A1/A2");
+ Path renameJson = new Path(path.getParent(), path.getName() + SUFFIX);
+ fs.mkdirs(renameJson);
+
+ fs.create(new Path(path.getParent(), "file1.txt"));
+ fs.create(new Path(path, "file2.txt"));
+
+ AtomicInteger redoRenameCall = new AtomicInteger(0);
+ Mockito.doAnswer(answer -> {
+ redoRenameCall.incrementAndGet();
+ return answer.callRealMethod();
+ }).when(client).getRedoRenameAtomicity(Mockito.any(Path.class),
+ Mockito.anyInt(), Mockito.any(TracingContext.class));
+
+ FileStatus[] fileStatuses = fs.listStatus(path.getParent());
+
+ Assertions.assertThat(fileStatuses.length)
+ .describedAs("List should return 3 files")
+ .isEqualTo(3);
+
+ Assertions.assertThat(redoRenameCall.get())
+ .describedAs("No redo rename call should be made")
+ .isEqualTo(0);
+
+ Assertions.assertThat(
+ Arrays.stream(fileStatuses)
+ .anyMatch(status ->
renameJson.toUri().getPath().equals(status.getPath().toUri().getPath())))
+ .describedAs("Directory with suffix -RenamePending.json should
exist.")
+ .isTrue();
+ }
+ }
+
+ /**
+ * Test case to verify crash recovery during listing with multiple pending
rename JSON files.
+ *
+ * This test simulates a scenario where multiple pending rename JSON files
exist, ensuring that
+ * crash recovery properly handles the situation. It verifies that two redo
rename calls are made
+ * and that the list operation returns the correct number of paths.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void listCrashRecoveryWithMultipleJsonFile() throws Exception {
+ AzureBlobFileSystem fs = null;
+ try {
+ Path path = new Path("/hbase/A1/A2");
+
+ // 1st Json file
+ Path renameJson = new Path(path.getParent(), path.getName() + SUFFIX);
+ fs = createJsonFile(path, renameJson);
+ AbfsBlobClient client = (AbfsBlobClient) addSpyHooksOnClient(fs);
+
+ // 2nd Json file
+ Path path2 = new Path("/hbase/A1/A3");
+ fs.create(new Path(path2, "file3.txt"));
+
+ Path renameJson2 = new Path(path2.getParent(), path2.getName() + SUFFIX);
+ AzureBlobFileSystemStore.VersionedFileStatus fileStatus =
(AzureBlobFileSystemStore.VersionedFileStatus) fs.getFileStatus(path2);
+
+ new RenameAtomicity(path2, new Path("/hbase/test4"), renameJson2,
getTestTracingContext(fs, true), fileStatus.getEtag(), client).preRename();
+
+ fs.create(new Path(path, "file2.txt"));
+
+ AtomicInteger redoRenameCall = new AtomicInteger(0);
+ Mockito.doAnswer(answer -> {
+ redoRenameCall.incrementAndGet();
+ return answer.callRealMethod();
+ }).when(client).getRedoRenameAtomicity(Mockito.any(Path.class),
+ Mockito.anyInt(), Mockito.any(TracingContext.class));
+
+ FileStatus[] fileStatuses = fs.listStatus(path.getParent());
+
+ Assertions.assertThat(fileStatuses.length)
+ .describedAs("List should return 2 paths")
+ .isEqualTo(2);
+
+ Assertions.assertThat(redoRenameCall.get())
+ .describedAs("2 redo rename calls should be made")
+ .isEqualTo(2);
+ } finally {
+ if (fs != null) {
+ fs.close();
+ }
+ }
+ }
+
+ /**
+ * Test case to verify path status when a pending rename JSON file exists.
+ *
+ * This test simulates a scenario where a rename operation was pending, and
ensures that
+ * the path status retrieval triggers a redo rename operation. The test also
checks that
+ * the correct error code (`PATH_NOT_FOUND`) is returned.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void getPathStatusWithPendingJsonFile() throws Exception {
+ AzureBlobFileSystem fs = null;
+ try {
+ Path path = new Path("/hbase/A1/A2");
+ Path renameJson = new Path(path.getParent(), path.getName() + SUFFIX);
+ fs = createJsonFile(path, renameJson);
+
+ AbfsBlobClient client = (AbfsBlobClient) addSpyHooksOnClient(fs);
+
+ fs.create(new Path("/hbase/A1/file1.txt"));
+ fs.create(new Path("/hbase/A1/file2.txt"));
+
+ AbfsConfiguration conf = fs.getAbfsStore().getAbfsConfiguration();
+
+ AtomicInteger redoRenameCall = new AtomicInteger(0);
+ Mockito.doAnswer(answer -> {
+ redoRenameCall.incrementAndGet();
+ return answer.callRealMethod();
+ }).when(client).getRedoRenameAtomicity(Mockito.any(Path.class),
+ Mockito.anyInt(), Mockito.any(TracingContext.class));
+
+ TracingContext tracingContext = new TracingContext(
+ conf.getClientCorrelationId(), fs.getFileSystemId(),
+ FSOperationType.GET_FILESTATUS, TracingHeaderFormat.ALL_ID_FORMAT,
null);
+
+ AzureServiceErrorCode azureServiceErrorCode = intercept(
+ AbfsRestOperationException.class, () -> client.getPathStatus(
+ path.toUri().getPath(), true,
+ tracingContext, null)).getErrorCode();
+
+ Assertions.assertThat(azureServiceErrorCode.getErrorCode())
+ .describedAs("Path had to be recovered from atomic rename
operation.")
+ .isEqualTo(PATH_NOT_FOUND.getErrorCode());
+
+ Assertions.assertThat(redoRenameCall.get())
+ .describedAs("There should be one redo rename call")
+ .isEqualTo(1);
+ } finally {
+ if (fs != null) {
+ fs.close();
+ }
+ }
+ }
+
+ /**
+ * Test case to verify path status when there is no pending rename JSON file.
+ *
+ * This test ensures that when no rename pending JSON file is present, the
path status is
+ * successfully retrieved, the ETag is present, and no redo rename operation
is triggered.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void getPathStatusWithoutPendingJsonFile() throws Exception {
+ try (AzureBlobFileSystem fs = Mockito.spy(this.getFileSystem())) {
+ assumeBlobServiceType();
+
+ Path path = new Path("/hbase/A1/A2");
+ AbfsBlobClient client = (AbfsBlobClient) addSpyHooksOnClient(fs);
+
+ fs.create(new Path(path, "file1.txt"));
+ fs.create(new Path(path, "file2.txt"));
+
+ AbfsConfiguration conf = fs.getAbfsStore().getAbfsConfiguration();
+
+ AtomicInteger redoRenameCall = new AtomicInteger(0);
+ Mockito.doAnswer(answer -> {
+ redoRenameCall.incrementAndGet();
+ return answer.callRealMethod();
+ }).when(client).getRedoRenameAtomicity(
+ Mockito.any(Path.class), Mockito.anyInt(),
+ Mockito.any(TracingContext.class));
+
+ TracingContext tracingContext = new TracingContext(
+ conf.getClientCorrelationId(), fs.getFileSystemId(),
+ FSOperationType.GET_FILESTATUS, TracingHeaderFormat.ALL_ID_FORMAT,
+ null);
+
+ AbfsHttpOperation abfsHttpOperation = client.getPathStatus(
+ path.toUri().getPath(), true,
+ tracingContext, null).getResult();
+
+ Assertions.assertThat(abfsHttpOperation.getStatusCode())
+ .describedAs("Path should be found.")
+ .isEqualTo(HTTP_OK);
+
+ Assertions.assertThat(extractEtagHeader(abfsHttpOperation))
+ .describedAs("Etag should be present.")
+ .isNotNull();
+
+ Assertions.assertThat(redoRenameCall.get())
+ .describedAs("There should be no redo rename call.")
+ .isEqualTo(0);
+ }
+ }
+
+ /**
+ * Test case to verify path status when there is a pending rename JSON
directory.
+ *
+ * This test simulates the scenario where a directory is created with a
rename pending JSON
+ * file (indicated by a specific suffix). It ensures that the path is found,
the ETag is present,
+ * and no redo rename operation is triggered. It also verifies that the
rename pending directory
+ * exists.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void getPathStatusWithPendingJsonDir() throws Exception {
+ try (AzureBlobFileSystem fs = Mockito.spy(this.getFileSystem())) {
+ assumeBlobServiceType();
+
+ Path path = new Path("/hbase/A1/A2");
+ AbfsBlobClient client = (AbfsBlobClient) addSpyHooksOnClient(fs);
+
+ fs.create(new Path(path, "file1.txt"));
+ fs.create(new Path(path, "file2.txt"));
+
+ fs.mkdirs(new Path(path.getParent(), path.getName() + SUFFIX));
+
+ AbfsConfiguration conf = fs.getAbfsStore().getAbfsConfiguration();
+
+ AtomicInteger redoRenameCall = new AtomicInteger(0);
+ Mockito.doAnswer(answer -> {
+ redoRenameCall.incrementAndGet();
+ return answer.callRealMethod();
+ }).when(client).getRedoRenameAtomicity(Mockito.any(Path.class),
+ Mockito.anyInt(), Mockito.any(TracingContext.class));
+
+ TracingContext tracingContext = new TracingContext(
+ conf.getClientCorrelationId(), fs.getFileSystemId(),
+ FSOperationType.GET_FILESTATUS, TracingHeaderFormat.ALL_ID_FORMAT,
null);
+
+ AbfsHttpOperation abfsHttpOperation =
client.getPathStatus(path.toUri().getPath(), true, tracingContext,
null).getResult();
+
+ Assertions.assertThat(abfsHttpOperation.getStatusCode())
+ .describedAs("Path should be found.")
+ .isEqualTo(HTTP_OK);
+
+ Assertions.assertThat(extractEtagHeader(abfsHttpOperation))
+ .describedAs("Etag should be present.")
+ .isNotNull();
+
+ Assertions.assertThat(redoRenameCall.get())
+ .describedAs("There should be no redo rename call.")
+ .isEqualTo(0);
+
+ Assertions.assertThat(fs.exists(new Path(path.getParent(),
path.getName() + SUFFIX)))
+ .describedAs("Directory with suffix -RenamePending.json should
exist.")
+ .isTrue();
+ }
+ }
+
+ /**
+ * Test case to verify the behavior when the ETag of a file changes during a
rename operation.
+ *
+ * This test simulates a scenario where the ETag of a file changes after the
creation of a
+ * rename pending JSON file. The steps include:
+ * - Creating a rename pending JSON file with an old ETag.
+ * - Deleting the original directory for an ETag change.
+ * - Creating new files in the directory.
+ * - Verifying that the copy blob call is not triggered.
+ * - Verifying that the rename atomicity operation is called once.
+ *
+ * The test ensures that the system correctly handles the ETag change during
the rename process.
+ *
+ * @throws Exception if any error occurs during the test execution
+ */
+ @Test
+ public void eTagChangedDuringRename() throws Exception {
Review Comment:
Added test wherever required.
##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRename.java:
##########
@@ -1655,6 +1730,827 @@ public void
testRenameSrcDirDeleteEmitDeletionCountInClientRequestId()
fs.rename(new Path(dirPathStr), new Path("/dst/"));
}
+ /**
+ * Helper method to configure the AzureBlobFileSystem and rename directories.
+ *
+ * @param currentFs The current AzureBlobFileSystem to use for renaming.
+ * @param producerQueueSize Maximum size of the producer queue.
+ * @param consumerMaxLag Maximum lag allowed for the consumer.
+ * @param maxThread Maximum threads for the rename operation.
+ * @param src The source path of the directory to rename.
+ * @param dst The destination path of the renamed directory.
+ * @throws IOException If an I/O error occurs during the operation.
+ */
+ private void renameDir(AzureBlobFileSystem currentFs, String
producerQueueSize,
+ String consumerMaxLag, String maxThread, Path src, Path dst)
+ throws IOException {
+ Configuration config = createConfig(producerQueueSize, consumerMaxLag,
maxThread);
+ try (AzureBlobFileSystem fs = (AzureBlobFileSystem)
FileSystem.newInstance(currentFs.getUri(), config)) {
+ fs.rename(src, dst);
+ validateRename(fs, src, dst, false, true, false);
+ }
+ }
+
+ /**
+ * Helper method to create the configuration for the AzureBlobFileSystem.
+ *
+ * @param producerQueueSize Maximum size of the producer queue.
+ * @param consumerMaxLag Maximum lag allowed for the consumer.
+ * @param maxThread Maximum threads for the rename operation.
+ * @return The configuration object.
+ */
+ private Configuration createConfig(String producerQueueSize, String
consumerMaxLag, String maxThread) {
+ Configuration config = new Configuration(this.getRawConfiguration());
+ config.set(FS_AZURE_PRODUCER_QUEUE_MAX_SIZE, producerQueueSize);
+ config.set(FS_AZURE_CONSUMER_MAX_LAG, consumerMaxLag);
+ config.set(FS_AZURE_BLOB_DIR_RENAME_MAX_THREAD, maxThread);
+ return config;
+ }
+
+ /**
+ * Helper method to validate that the rename was successful and that the
destination exists.
+ *
+ * @param fs The AzureBlobFileSystem instance to check the existence on.
+ * @param dst The destination path.
+ * @param src The source path.
+ * @throws IOException If an I/O error occurs during the validation.
+ */
+ private void validateRename(AzureBlobFileSystem fs, Path src, Path dst,
+ boolean isSrcExist, boolean isDstExist, boolean isJsonExist)
+ throws IOException {
+ Assertions.assertThat(fs.exists(dst))
+ .describedAs("Renamed Destination directory should exist.")
+ .isEqualTo(isDstExist);
+ Assertions.assertThat(fs.exists(new Path(src.getParent(), src.getName() +
SUFFIX)))
+ .describedAs("Renamed Pending Json file should exist.")
+ .isEqualTo(isJsonExist);
+ Assertions.assertThat(fs.exists(src))
+ .describedAs("Renamed Destination directory should exist.")
+ .isEqualTo(isSrcExist);
+ }
+
+ /**
+ * Test the renaming of a directory with different parallelism
configurations.
+ */
+ @Test
+ public void testRenameDirWithDifferentParallelism() throws Exception {
Review Comment:
Taken!
> ABFS: [FnsOverBlob][Tests] Add Tests For Negative Scenarios Identified for
> Rename Operation
> -------------------------------------------------------------------------------------------
>
> Key: HADOOP-19445
> URL: https://issues.apache.org/jira/browse/HADOOP-19445
> Project: Hadoop Common
> Issue Type: Sub-task
> Components: fs/azure
> Affects Versions: 3.4.1
> Reporter: Anuj Modi
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
> Attachments: RenameFile_TestScenarios.pdf,
> RenameFolder_TestScenarios.pdf
>
>
> We have identified a few scenarios worth adding integration or mocked
> behavior tests for while implementing FNS Support over Blob Endpoint.
> Attached file shows the the scenarios identified for Rename File and Rename
> directory operations on blob Endpoint
> This Jira tracks implementing these tests.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
