AbfsoutputStream refactor #118

anmolanmol1234 · 2024-06-13T05:30:03Z

Description of PR :

This Pr is in correlation to the series of work done under Parent Jira: [HADOOP-19179]

(https://issues.apache.org/jira/browse/HADOOP-19179)
Jira for this Patch: https://issues.apache.org/jira/browse/HADOOP-19187

Scope of this task is to refactor the AbfsOutputStream class to handle the ingress for DFS and Blob endpoint effectively.

Production code changes :

The AbfsOutputStream class is crucial for handling the data being written to Azure Storage. Its primary responsibilities include:

Buffering: Temporarily holding data in memory before it is uploaded.
Streaming: Efficiently streaming data to Azure Storage.
Committing: Ensuring that buffered data is correctly uploaded and committed to Azure Storage.

New Additions

The new additions introduce a more modular and flexible approach to managing data ingress (data being written to storage), catering to both Azure Data Lake Storage (ADLS) and Azure Blob Storage.

AzureIngressHandler

The AzureIngressHandler is a new parent class designed to encapsulate common logic for data ingress operations. It simplifies the process of writing data to Azure Storage by providing a unified interface. This class has two specialized child classes:

AzureDfsIngressHandler:
- Manages data ingress specifically for Azure Data Lake Storage (DFS).
- Handles operations like creating, appending, and flushing data blocks for DFS.
AzureBlobIngressHandler:
- Manages data ingress specifically for Azure Blob Storage (BLOB).
- Handles operations like creating, appending, and flushing data blocks for Blob Storage, while ensuring that each block has a unique blockId.

AbfsBlock and AbfsBlobBlock

Data is managed in discrete blocks to improve efficiency and manageability.

AbfsBlock:
- A basic structure for buffering data.
- Used as a common block type for both DFS and Blob Storage.
AbfsBlobBlock:
- A subclass of AbfsBlock tailored for Blob Storage.
- Requires a unique blockId for each block, which is necessary for the Blob Storage API.

Block Managers

To manage these data blocks, new manager classes have been introduced. These classes handle the lifecycle of blocks, including creation, appending, and flushing.

AzureBlockManager:
- A parent class for managing the lifecycle of data blocks.
- Provides common functionality for block management.
AzureDFSBlockManager:
- Manages the lifecycle of AbfsBlock instances for DFS.
- Handles the specifics of appending and flushing blocks in DFS.
AzureBlobBlockManager:
- Manages the lifecycle of AbfsBlobBlock instances for Blob Storage.
- Ensures each block has a unique blockId.
- Handles the specifics of appending and flushing blocks in Blob Storage.

Integration with AbfsOutputStream

The AbfsOutputStream class has been updated to incorporate the new ingress flow logic, enhancing its ability to handle data writes to both DFS and Blob Storage. Here’s how it integrates:

Configuration Selection:
- The AbfsOutputStream reads the configuration parameter fs.azure.ingress.service.type to determine whether the user has configured the system to use BLOB or DFS for data ingress.
Handler Initialization:
- Based on the configuration, AbfsOutputStream initializes the appropriate handler (AzureBlobIngressHandler or AzureDfsIngressHandler).
Buffering Data:
- As data is written to AbfsOutputStream, it is buffered into blocks (AbfsBlock for DFS or AbfsBlobBlock for Blob Storage).
Managing Blocks:
- The corresponding block manager (AzureDFSBlockManager or AzureBlobBlockManager) manages the lifecycle of these blocks, ensuring that data is correctly created, appended, and flushed.
Block Id Management (Blob Specific):
- For Blob Storage, AzureBlobBlockManager ensures that each block has a unique blockId, adhering to the requirements of the Blob Storage API.

Detailed Flow

Creating Data Blocks:
- When data is written to AbfsOutputStream, it is divided into blocks (AbfsBlock for DFS or AbfsBlobBlock for Blob Storage).
Appending Data:
- These blocks are appended to the Azure storage system via the appropriate handler (AzureBlobIngressHandler or AzureDfsIngressHandler).
Flushing Data:
- Once all data has been buffered, the blocks are flushed to ensure all buffered data is committed to the storage system.
Lifecycle Management:
- The block managers (AzureDFSBlockManager and AzureBlobBlockManager) oversee the lifecycle of blocks, handling retries, errors, and ensuring data integrity.

Test Code Changes:

Existing tests are modified to work with new abstracted design.
Test Suite was run on DFS Endpoint to make sure the original driver works seamlessly and undisturbed.
Adding some new tests around the new code

anujmodi2021

Added some comments.

...tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java

...ls/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java

...op-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java

...p-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlobBlock.java

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlock.java

...tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlockStatus.java

…adoop into azureBlobClientOriginal

saxenapranav

Thanks for taking the suggestions. +1

anujmodi2021

LGTM...
Thanks

...ls/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java

saxenapranav

Great code! minor comments. One suggestion related to the outputStream verification in #prepareListToCommit.

saxenapranav · 2024-06-24T08:50:01Z

...ls/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java

+    APR_10_2021("2021-04-10"),
    AUG_06_2021("2021-08-06"),


Aug06 looks unused, lets remove it.

saxenapranav · 2024-06-24T08:50:35Z

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java

@@ -109,6 +109,7 @@ public abstract class AbfsClient implements Closeable {

  private final URL baseUrl;
  private final SharedKeyCredentials sharedKeyCredentials;
+  // TODO: Abstract for blob and dfs for CPK in OSS PR


TODO comment.

Will take care of it while raising OSS PR. Currently we are planning to do scale tests with the latest version, so added a TODO as a reminder

saxenapranav · 2024-06-24T08:52:48Z

...s/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AzureIngressHandler.java

-        .contains(BLOB_OPERATION_NOT_SUPPORTED);
+    return ex.getStatusCode() == HTTP_CONFLICT && (ex.getErrorCode()
+        .getErrorCode().equals(AzureServiceErrorCode.BLOB_OPERATION_NOT_SUPPORTED.getErrorCode()) ||
+        ex.getErrorMessage().contains(INVALID_APPEND_OPERATION));


can we take dependency on error code and not on the message (message might change in future, or if that is not the case)

Have written a mail to XFE folks to confirm that can we take dependency on this error code because InvalidAppend is a very generic one. We want to confirm if there are any other scenarios where they return this error code. Once they confirm will update the code as needed

saxenapranav · 2024-06-24T09:20:35Z

...c/main/java/org/apache/hadoop/fs/azurebfs/services/AzureDfsToBlobIngressFallbackHandler.java

+    if (blockManager instanceof AzureBlobBlockManager) {
+      this.blobBlockManager = (AzureBlobBlockManager) blockManager;
+    } else {


looks like this is redundant check. reason being, this class can be used only if the outputStream is a newly created class. From the code of switchHadler, looks like this class cant come in switching.

saxenapranav · 2024-06-24T09:22:22Z

...adoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AzureDFSIngressHandler.java

+    if (blockManager instanceof AzureDFSBlockManager) {
+      this.dfsBlockManager = (AzureDFSBlockManager) blockManager;
+    } else {


This also looks a redundant check, as for the two cases when this class would inited:

outputStream init: blockManager would be null.

in switch, the blockManager sent would be blobBlockManager.

saxenapranav · 2024-06-24T09:39:59Z

...ools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java

      if (switchCompleted && ingressHandler != null) {
        return ingressHandler; // Return the existing ingress handler
      }
-      this.client = clientHandler.getClient(serviceType);
-      if (serviceType == AbfsServiceType.BLOB) {
+      if (isDFSToBlobFallbackEnabled && serviceTypeAtInit == AbfsServiceType.BLOB) {


Should we add a constraint that there can not be switch from dfsHandler to blobHandler if the config is not given. As in if the dfs appends error for inter-op, and the config is not given. Job should throw error.

This happens already because we don't have switchHandler handling in case of DFS appends as they don't throw the InvalidIngressServiceException, and we switch only if that exception is caught. So DFSIngressHandler to BlobIngress is not posssible via code

saxenapranav · 2024-06-24T09:48:10Z

...ools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java

+      } catch (InvalidIngressServiceException ex) {
+        // If an invalid ingress service is encountered, switch handler and retry
+        switchHandler();
+        op = remoteFlush(offset, retainUncommittedData, isClose, leaseId,
+            tracingContext);


Dont think that switch can happen at this this state. Or are you referring that the outputStream is closed without writing anything? If yes, then probably we can completely leave flushing if no write has happened? What you say?

So DFS today calls flush even if there are no appends, in case of Blob handling we have taken care of this by not doing any PutBlockList calls if map is empty. Since we are planning not to change the DFS behaviour, this handling is needed.

saxenapranav · 2024-06-24T10:13:31Z

...hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AzureBlobBlockManager.java

+
+    int mapEntry = 0;
+    // If any of the entry in the map doesn't have the status of SUCCESS, fail the flush.
+    for (Map.Entry<String, AbfsBlockStatus> entry : getBlockStatusMap().entrySet()) {


I believe that this block is to verify if the AbfsOutputStream has correctly added blockIds in correct sequence. Is there any more reason?

If its to verify, I would be more inclined towards removing this block -> this would save computation at flush + computation to set blockStatusMap. And as far as the verification is concerned, I would say our integ tests should give that confidence (plus if this is breaking, this will break at large scale).

If we remove this block, I would suggest further to remove blockStatusMap totally. orderedBlockList is the important data-structure for us.

Appreciate the suggestion here but would like to keep this code block as it is. Computation cost is not much here.

I believe that this block is to verify if the AbfsOutputStream has correctly added blockIds in correct sequence. Is there any more reason? -> this is not just for sequence, it validates that all appends are successful before flushing

This code would be firing at flushWrittenBytesToServiceInternal which has two callers:

flushWrittenBytesToServiceAsync

flushWrittenBytesToService
In both the cases, flushWrittenBytesToServiceInternal can be only called when the parallel threads are completely done sucessfully.

So, technically this flush api on server would be getting called only once the writing is successful on all the threads. As per code, the data-consistency for write / flush is maintained.

anmolanmol1234 added 30 commits May 22, 2024 04:47

AbfsHandlers

15fdd85

AbfsOutputStream changes

2e019a9

Synchronization issues

6969b23

Small write

1942e61

Javadocs update

2adae79

Javadocs

f653b07

Javadocs

ea5b71e

appendblob

e9d93dd

merge conflict

095bc4b

length 0

48e5b8f

close AbfsBlock

da83b8a

add new error

4d2dc86

Add ingress handler to tracing context

20002d5

logging

54535fd

Add AppendBlob support for FNS

75fd3a1

close data block for append blob

6dba239

Add marker blob creation logic

bbe2230

changes for create

96b9f46

extra line

a84207a

Append fixes

0e373c5

parent changes

450dc14

Back merge

2520979

merge conflicts

f0bac54

Client orchestration

39790f6

test fix

2e3a9e4

test fixes

68ca484

test fix

eb1e655

Fix for switching handler

c39c389

final type

f1318e6

tests

b54cf97

anmolanmol1234 added 7 commits June 18, 2024 00:16

create flow fix

2fff127

fix recursion

a847923

fix call

63c060b

fix test

513aa52

review comments

f1c6261

createBlock synchronized

7c24072

missed file

b6e3f77

anujmodi2021 reviewed Jun 19, 2024

View reviewed changes

anmolanmol1234 added 3 commits June 19, 2024 03:07

review comments

9f03a9b

Merge branch 'azureBlobClient' of https://github.com/ABFSDriver/AbfsH…

43e7da1

…adoop into azureBlobClientOriginal

take care of switching only once

f0c233f

saxenapranav approved these changes Jun 19, 2024

View reviewed changes

anmolanmol1234 changed the base branch from azureBlobClient to wasbDepCodeReview June 19, 2024 10:53

anujmodi2021 approved these changes Jun 19, 2024

View reviewed changes

remove small write check

327fb1f

anujmodi2021 requested changes Jun 20, 2024

View reviewed changes

...ls/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java Outdated Show resolved Hide resolved

anujmodi2021 approved these changes Jun 20, 2024

View reviewed changes

anmolanmol1234 added 6 commits June 20, 2024 02:47

Handle fallback effectively

e4ce98c

missed file

3b087dc

change to getter

ff8e959

fallback for appendblob

f658a68

correct try catch

925fcf0

commit for metrics test

887fa4d

saxenapranav requested changes Jun 24, 2024

View reviewed changes

anmolanmol1234 added 4 commits June 24, 2024 04:30

review comments

b41b3d9

changes for block manager

942ff29

no param needed

4fce8b4

revert no param

1414b60

saxenapranav approved these changes Jun 24, 2024

View reviewed changes

anmolanmol1234 merged commit 8937ae5 into wasbDepCodeReview Jun 25, 2024
1 check passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

AbfsoutputStream refactor #118

AbfsoutputStream refactor #118

anmolanmol1234 commented Jun 13, 2024 •

edited

Loading

anujmodi2021 left a comment

saxenapranav left a comment

anujmodi2021 left a comment

saxenapranav left a comment

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024 •

edited

Loading

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024 •

edited

Loading

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024

saxenapranav Jun 24, 2024

saxenapranav Jun 24, 2024

anmolanmol1234 Jun 24, 2024

anmolanmol1234 Jun 24, 2024

saxenapranav Jun 24, 2024

AbfsoutputStream refactor #118

AbfsoutputStream refactor #118

Conversation

anmolanmol1234 commented Jun 13, 2024 • edited Loading

Description of PR :

Production code changes :

New Additions

AzureIngressHandler

AbfsBlock and AbfsBlobBlock

Block Managers

Integration with AbfsOutputStream

Detailed Flow

Test Code Changes:

anujmodi2021 left a comment

Choose a reason for hiding this comment

saxenapranav left a comment

Choose a reason for hiding this comment

anujmodi2021 left a comment

Choose a reason for hiding this comment

saxenapranav left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

anmolanmol1234 Jun 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

anmolanmol1234 Jun 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

anmolanmol1234 commented Jun 13, 2024 •

edited

Loading

anmolanmol1234 Jun 24, 2024 •

edited

Loading

anmolanmol1234 Jun 24, 2024 •

edited

Loading